attempt to Protect scenario custom properties

taipy/core/_entity/_properties.py

@@ -12,7 +12,9 @@
 from collections import UserDict
 
 from taipy.config.common._template_handler import _TemplateHandler as _tpl
+from taipy.config.common._validate_id import _validate_id
 
+from ..exceptions.exceptions import PropertyKeyAlreadyExisted
 from ..notification import EventOperation, Notifier, _make_event
 
 
@@ -26,11 +28,16 @@ class _Properties(UserDict):
         self._pending_deletions = set()
 
     def __setitem__(self, key, value):
-        super(_Properties, self).__setitem__(key, value)
-
         if hasattr(self, "_entity_owner"):
             from ... import core as tp
 
+            entity_owner = tp.get(self._entity_owner)
+            try:
+                entity_owner._get_attributes(_validate_id(key), key)
+                raise PropertyKeyAlreadyExisted(key)
+            except AttributeError:
+                super(_Properties, self).__setitem__(key, value)
+
             event = _make_event(
                 self._entity_owner,
                 EventOperation.UPDATE,
@@ -45,6 +52,8 @@ class _Properties(UserDict):
                     self._pending_deletions.remove(key)
                 self._pending_changes[key] = value
                 self._entity_owner._in_context_attributes_changed_collector.append(event)
+        else:
+            super(_Properties, self).__setitem__(key, value)
 
     def __getitem__(self, key):
         return _tpl._replace_templates(super(_Properties, self).__getitem__(key))

taipy/core/exceptions/exceptions.py

@@ -383,3 +383,17 @@ class SQLQueryCannotBeExecuted(Exception):
 
 class _SuspiciousFileOperation(Exception):
     pass
+
+
+class AttributeKeyAlreadyExisted(Exception):
+    """Raised when an attribute key already existed."""
+
+    def __init__(self, key: str):
+        self.message = f"Attribute key '{key}' already existed."
+
+
+class PropertyKeyAlreadyExisted(Exception):
+    """Raised when a property key already existed."""
+
+    def __init__(self, key: str):
+        self.message = f"Property key '{key}' already existed."

taipy/core/scenario/scenario.py

@@ -31,6 +31,7 @@ from ..cycle.cycle import Cycle
 from ..data.data_node import DataNode
 from ..data.data_node_id import DataNodeId
 from ..exceptions.exceptions import (
+    AttributeKeyAlreadyExisted,
     InvalidSequence,
     NonExistingDataNode,
     NonExistingSequence,
@@ -176,11 +177,40 @@ class Scenario(_Entity, Submittable, _Labeled):
     def __eq__(self, other):
         return isinstance(other, Scenario) and self.id == other.id
 
-    def __getattr__(self, attribute_name):
-        protected_attribute_name = _validate_id(attribute_name)
-        if protected_attribute_name in self._properties:
-            return _tpl._replace_templates(self._properties[protected_attribute_name])
-
+    def __setattr__(self, name: str, value: Any) -> None:
+        entity_taipy_attributes = [
+            "_config_id",
+            "id",
+            "_tasks",
+            "_additional_data_nodes",
+            "_creation_date",
+            "_cycle",
+            "_primary_scenario",
+            "_tags",
+            "_properties",
+            "_sequences",
+            "_version",
+            "_submittable_id",
+            "_subscribers",  # from Submittable class
+            "_in_context_attributes_changed_collector",
+            "_is_in_context"  # from _Entity class
+            "name",  # from def name setter
+        ]
+        if name in entity_taipy_attributes:
+            return super().__setattr__(name, value)
+        else:
+            protected_attribute_name = _validate_id(name)
+            try:
+                if protected_attribute_name not in self._properties and not self._get_attributes(
+                    protected_attribute_name, name
+                ):
+                    raise AttributeError  #   throw AttributeError if found
+
+                raise AttributeKeyAlreadyExisted(name)
+            except AttributeError:
+                return super().__setattr__(name, value)
+
+    def _get_attributes(self, protected_attribute_name, attribute_name):
         sequences = self._get_sequences()
         if protected_attribute_name in sequences:
             return sequences[protected_attribute_name]
@@ -190,8 +220,17 @@ class Scenario(_Entity, Submittable, _Labeled):
         data_nodes = self.data_nodes
         if protected_attribute_name in data_nodes:
             return data_nodes[protected_attribute_name]
+
         raise AttributeError(f"{attribute_name} is not an attribute of scenario {self.id}")
 
+    def __getattr__(self, attribute_name):
+        protected_attribute_name = _validate_id(attribute_name)
+        if hasattr(self, "_properties"):
+            if protected_attribute_name in self._properties:
+                return _tpl._replace_templates(self._properties[protected_attribute_name])
+
+        return self._get_attributes(protected_attribute_name, attribute_name)
+
     @property
     def config_id(self):
         return self._config_id

tests/core/scenario/test_scenario.py

@@ -15,14 +15,21 @@ import pytest
 
 from taipy.config import Frequency
 from taipy.config.common.scope import Scope
+from taipy.config.config import Config
 from taipy.config.exceptions.exceptions import InvalidConfigurationId
+from taipy.core import create_scenario
 from taipy.core.common._utils import _Subscriber
 from taipy.core.cycle._cycle_manager_factory import _CycleManagerFactory
 from taipy.core.cycle.cycle import Cycle, CycleId
 from taipy.core.data._data_manager_factory import _DataManagerFactory
 from taipy.core.data.in_memory import DataNode, InMemoryDataNode
 from taipy.core.data.pickle import PickleDataNode
-from taipy.core.exceptions.exceptions import SequenceAlreadyExists, SequenceTaskDoesNotExistInScenario
+from taipy.core.exceptions.exceptions import (
+    AttributeKeyAlreadyExisted,
+    PropertyKeyAlreadyExisted,
+    SequenceAlreadyExists,
+    SequenceTaskDoesNotExistInScenario,
+)
 from taipy.core.scenario._scenario_manager_factory import _ScenarioManagerFactory
 from taipy.core.scenario.scenario import Scenario
 from taipy.core.scenario.scenario_id import ScenarioId
@@ -156,6 +163,29 @@ def test_create_scenario_and_add_sequences():
     assert scenario.sequences == {"sequence_1": scenario.sequence_1, "sequence_2": scenario.sequence_2}
 
 
+def test_get_set_property_to_scenario():
+    dn_cfg = Config.configure_data_node("bar")
+    s_cfg = Config.configure_scenario("foo", additional_data_node_configs=[dn_cfg], key="value")
+    scenario = create_scenario(s_cfg)
+
+    assert scenario.properties == {"key": "value"}
+    assert scenario.key == "value"
+
+    scenario.properties["new_key"] = "new_value"
+    scenario.another_key = "another_value"
+
+    assert scenario.key == "value"
+    assert scenario.new_key == "new_value"
+    assert scenario.another_key == "another_value"
+    assert scenario.properties == {"key": "value", "new_key": "new_value"}
+
+    with pytest.raises(AttributeKeyAlreadyExisted):
+        scenario.bar = "KeyAlreadyUsed"
+
+    with pytest.raises(PropertyKeyAlreadyExisted):
+        scenario.properties["bar"] = "KeyAlreadyUsed"
+
+
 def test_create_scenario_overlapping_sequences():
     input_1 = PickleDataNode("input_1", Scope.SCENARIO)
     output_1 = PickleDataNode("output_1", Scope.SCENARIO)