feat: add can_create api

taipy/core/data/_data_manager.py

@@ -24,6 +24,8 @@ from ..config.data_node_config import DataNodeConfig
 from ..cycle.cycle_id import CycleId
 from ..exceptions.exceptions import InvalidDataNodeType
 from ..notification import Event, EventEntityType, EventOperation, Notifier, _make_event
+from ..reason._reason_factory import _build_config_can_not_create_reason, _build_not_global_datanode_config_reason
+from ..reason.reason import Reasons
 from ..scenario.scenario_id import ScenarioId
 from ..sequence.sequence_id import SequenceId
 from ._data_fs_repository import _DataFSRepository
@@ -68,6 +70,18 @@ class _DataManager(_Manager[DataNode], _VersionMixin):
             for dn_config, owner_id in dn_configs_and_owner_id
         }
 
+    @classmethod
+    def _can_create(cls, config: DataNodeConfig) -> Reasons:
+        config_id = getattr(config, "id", None) or str(config)
+        reason = Reasons(config_id)
+
+        if not isinstance(config, DataNodeConfig):
+            reason._add_reason(config_id, _build_config_can_not_create_reason(config_id))
+        elif config.scope is not Scope.GLOBAL:
+            reason._add_reason(config_id, _build_not_global_datanode_config_reason(config_id))
+
+        return reason
+
     @classmethod
     def _create_and_set(
         cls, data_node_config: DataNodeConfig, owner_id: Optional[str], parent_ids: Optional[Set[str]]

taipy/core/reason/_reason_factory.py

@@ -22,3 +22,11 @@ def _build_data_node_is_not_written(dn_id: DataNodeId) -> str:
 
 def _build_not_submittable_entity_reason(entity_id: str) -> str:
     return f"Entity {entity_id} is not a submittable entity"
+
+
+def _build_config_can_not_create_reason(config_id: str) -> str:
+    return f'Object "{config_id}" is not a valid config to be created'
+
+
+def _build_not_global_datanode_config_reason(config_id: str) -> str:
+    return f'Data node config "{config_id}" does not have GLOBAL scope'

taipy/core/scenario/_scenario_manager.py

@@ -46,7 +46,7 @@ from ..exceptions.exceptions import (
 from ..job._job_manager_factory import _JobManagerFactory
 from ..job.job import Job
 from ..notification import EventEntityType, EventOperation, Notifier, _make_event
-from ..reason._reason_factory import _build_not_submittable_entity_reason
+from ..reason._reason_factory import _build_config_can_not_create_reason, _build_not_submittable_entity_reason
 from ..reason.reason import Reasons
 from ..submission._submission_manager_factory import _SubmissionManagerFactory
 from ..submission.submission import Submission
@@ -114,6 +114,16 @@ class _ScenarioManager(_Manager[Scenario], _VersionMixin):
             _make_event(scenario, EventOperation.UPDATE, attribute_name="subscribers", attribute_value=params)
         )
 
+    @classmethod
+    def _can_create(cls, config: ScenarioConfig) -> Reasons:
+        config_id = getattr(config, "id", None) or str(config)
+        reason = Reasons(config_id)
+
+        if not isinstance(config, ScenarioConfig):
+            reason._add_reason(config_id, _build_config_can_not_create_reason(config_id))
+
+        return reason
+
     @classmethod
     def _create(
         cls,

taipy/core/taipy.py

@@ -51,7 +51,7 @@ from .exceptions.exceptions import (
 from .job._job_manager_factory import _JobManagerFactory
 from .job.job import Job
 from .job.job_id import JobId
-from .reason._reason_factory import _build_not_submittable_entity_reason
+from .reason._reason_factory import _build_config_can_not_create_reason, _build_not_submittable_entity_reason
 from .reason.reason import Reasons
 from .scenario._scenario_manager_factory import _ScenarioManagerFactory
 from .scenario.scenario import Scenario
@@ -873,6 +873,23 @@ def get_cycles() -> List[Cycle]:
     return _CycleManagerFactory._build_manager()._get_all()
 
 
+def can_create(config: Union[ScenarioConfig, DataNodeConfig]) -> Reasons:
+    """Indicate if a config can be created.
+
+    This function checks if the given scenario or data node config can be created.
+
+    Returns:
+        True if the given config can be created. False otherwise.
+    """
+    if isinstance(config, ScenarioConfig):
+        return _ScenarioManagerFactory._build_manager()._can_create(config)
+    elif isinstance(config, DataNodeConfig):
+        return _DataManagerFactory._build_manager()._can_create(config)
+    else:
+        config_id = getattr(config, "id", None) or str(config)
+        return Reasons(config_id)._add_reason(config_id, _build_config_can_not_create_reason(config_id))
+
+
 def create_scenario(
     config: ScenarioConfig,
     creation_date: Optional[datetime] = None,

tests/core/data/test_data_manager.py

@@ -45,6 +45,24 @@ class TestDataManager:
         assert dn.properties.get("foo") == "bar"
         assert dn.properties.get("baz") == "qux"
 
+    def test_can_create(self):
+        dn_config = Config.configure_in_memory_data_node("dn", 10, scope=Scope.SCENARIO)
+        global_dn_config = Config.configure_data_node(
+            id="global_dn", storage_type="in_memory", scope=Scope.GLOBAL, data=10
+        )
+
+        reasons = _DataManager._can_create(global_dn_config)
+        assert bool(reasons) is True
+        assert reasons._reasons == {}
+
+        reasons = _DataManager._can_create(dn_config)
+        assert bool(reasons) is False
+        assert reasons._reasons == {dn_config.id: {'Data node config "dn" does not have GLOBAL scope'}}
+
+        reasons = _DataManager._can_create(1)
+        assert bool(reasons) is False
+        assert reasons._reasons == {"1": {'Object "1" is not a valid config to be created'}}
+
     def test_create_data_node_with_name_provided(self):
         dn_config = Config.configure_data_node(id="dn", foo="bar", name="acb")
         dn = _DataManager._create_and_set(dn_config, None, None)

tests/core/scenario/test_scenario_manager.py

@@ -365,6 +365,29 @@ def test_create_and_delete_scenario():
     assert len(_ScenarioManager._get_all()) == 0
 
 
+def test_can_create():
+    dn_config = Config.configure_in_memory_data_node("dn", 10)
+    task_config = Config.configure_task("task", print, [dn_config])
+    scenario_config = Config.configure_scenario("sc", {task_config}, [], Frequency.DAILY)
+
+    reasons = _ScenarioManager._can_create(scenario_config)
+    assert bool(reasons) is True
+    assert reasons._reasons == {}
+    _ScenarioManager._create(scenario_config)
+
+    reasons = _ScenarioManager._can_create(task_config)
+    assert bool(reasons) is False
+    assert reasons._reasons == {task_config.id: {'Object "task" is not a valid config to be created'}}
+    with pytest.raises(AttributeError):
+        _ScenarioManager._create(task_config)
+
+    reasons = _ScenarioManager._can_create(1)
+    assert bool(reasons) is False
+    assert reasons._reasons == {"1": {'Object "1" is not a valid config to be created'}}
+    with pytest.raises(AttributeError):
+        _ScenarioManager._create(1)
+
+
 def test_is_deletable():
     assert len(_ScenarioManager._get_all()) == 0
     scenario_config = Config.configure_scenario("sc", None, None, Frequency.DAILY)

tests/core/test_taipy.py

@@ -677,6 +677,17 @@ class TestTaipy:
             tp.exists(cycle_id)
             mck.assert_called_once_with(cycle_id)
 
+    def test_can_create(self):
+        global_dn_config = Config.configure_in_memory_data_node("global_dn", 10, scope=Scope.GLOBAL)
+        dn_config = Config.configure_in_memory_data_node("dn", 10)
+        task_config = Config.configure_task("task", print, [dn_config])
+        scenario_config = Config.configure_scenario("sc", {task_config}, [], Frequency.DAILY)
+
+        assert tp.can_create(scenario_config)
+        assert tp.can_create(global_dn_config)
+        assert not tp.can_create(dn_config)
+        assert not tp.can_create("1")
+
     def test_create_global_data_node(self):
         dn_cfg_global = DataNodeConfig("id", "pickle", Scope.GLOBAL)
         dn_cfg_scenario = DataNodeConfig("id", "pickle", Scope.SCENARIO)