feat: data register supports dynamically loaded classes

Author: cuinixam

src/py_app_dev/core/data_registry.py

@@ -14,17 +14,22 @@ def __init__(self, data: Any, provider_name: str) -> None:
 
 class DataRegistry:
     def __init__(self) -> None:
-        # Registry to store data entries by type
-        self._registry: Dict[Type[Any], List[DataEntry]] = defaultdict(list)
+        # Registry to store data entries by type (fully qualified name as the key)
+        self._registry: Dict[str, List[DataEntry]] = defaultdict(list)
+
+    @staticmethod
+    def _get_type_name(obj_type: Type[Any]) -> str:
+        """Get the fully qualified name of the type."""
+        return f"{obj_type.__module__}.{obj_type.__name__}"
 
     def insert(self, data: Any, provider: str) -> None:
         """Registers a piece of information with the provider."""
-        self._registry[type(data)].append(DataEntry(data, provider))
+        self._registry[self._get_type_name(type(data))].append(DataEntry(data, provider))
 
     def find_data(self, data_type: Type[T]) -> List[T]:
         """Find all data of a given type."""
         return [entry.data for entry in self.find_entries(data_type)]
 
     def find_entries(self, data_type: Type[T]) -> List[DataEntry]:
         """Find all data entries of a given type. Each entry contains the data and the provider."""
-        return self._registry.get(data_type, [])
+        return self._registry.get(self._get_type_name(data_type), [])

tests/conftest.py

@@ -0,0 +1,27 @@
+import textwrap
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def my_python_file(tmp_path: Path) -> Path:
+    """Fixture to create a temporary Python file with a sample step."""
+    python_file = tmp_path / "my_python_file.py"
+    python_file.write_text(
+        textwrap.dedent(
+            """\
+            from typing import List
+            from pathlib import Path
+            from py_app_dev.core.pipeline import PipelineStep
+            class MyStep(PipelineStep):
+                def run(self) -> None:
+                    pass
+                def get_dependencies(self) -> List[Path]:
+                    pass
+                def get_outputs(self) -> List[Path]:
+                    pass
+            """
+        )
+    )
+    return python_file

tests/test_data_registry.py

@@ -1,15 +1,16 @@
-from dataclasses import dataclass
+from pathlib import Path
 
 from py_app_dev.core.data_registry import DataRegistry
+from py_app_dev.core.pipeline import PipelineLoader, PipelineStep, PipelineStepConfig
 
 
-class Provider1:
-    pass
+class SomeData:
+    def __init__(self, data: str):
+        self.data = data
 
 
-@dataclass
-class SomeData:
-    name: str
+class Provider1:
+    pass
 
 
 def test_insert_and_find_data():
@@ -19,7 +20,7 @@ def test_insert_and_find_data():
 
     string_data = registry.find_data(SomeData)
     assert len(string_data) == 1
-    assert string_data[0].name == "new data"
+    assert string_data[0].data == "new data"
 
     int_data = registry.find_data(int)
     assert int_data == [123]
@@ -44,3 +45,21 @@ def test_find_data_empty():
     registry = DataRegistry()
     data = registry.find_data(str)
     assert data == []
+
+
+def test_support_dynamically_loaded_classes(my_python_file: Path) -> None:
+    def load_step():
+        return PipelineLoader[PipelineStep]._load_steps(
+            "install",
+            [PipelineStepConfig(step="MyStep", file=str(my_python_file), config={"data": "value"})],
+            my_python_file.parent,
+        )[0]
+
+    step_class_1st_load = load_step()._class
+    step_class_2nd_load = load_step()._class
+    assert step_class_1st_load != step_class_2nd_load  # Different instances
+    registry = DataRegistry()
+    registry.insert(step_class_1st_load(), "provider1")
+    registry.insert(step_class_2nd_load(), "provider1")
+    data = registry.find_data(step_class_2nd_load)
+    assert len(data) == 2

tests/test_pipeline.py

@@ -1,4 +1,3 @@
-import textwrap
 from pathlib import Path
 from typing import Any, List, OrderedDict
 
@@ -8,29 +7,6 @@
 from py_app_dev.core.pipeline import PipelineLoader, PipelineStep, PipelineStepConfig
 
 
-@pytest.fixture
-def my_python_file(tmp_path: Path) -> Path:
-    """Fixture to create a temporary Python file with a sample step."""
-    python_file = tmp_path / "my_python_file.py"
-    python_file.write_text(
-        textwrap.dedent(
-            """\
-            from typing import List
-            from pathlib import Path
-            from py_app_dev.core.pipeline import PipelineStep
-            class MyStep(PipelineStep):
-                def run(self) -> None:
-                    pass
-                def get_dependencies(self) -> List[Path]:
-                    pass
-                def get_outputs(self) -> List[Path]:
-                    pass
-            """
-        )
-    )
-    return python_file
-
-
 def test_load_unknown_step():
     with pytest.raises(UserNotificationException):
         PipelineLoader[PipelineStep]._load_steps("install", [PipelineStepConfig(step="StepIDontExist")], Path("."))