Add support for registering with, and resolving by, tags.

Author: runemalm

src/dependency_injection/container.py

@@ -1,6 +1,6 @@
 import inspect
 
-from typing import Any, Dict, Optional, TypeVar, Type
+from typing import Any, Dict, List, Optional, TypeVar, Type
 
 from dependency_injection.registration import Registration
 from dependency_injection.scope import DEFAULT_SCOPE_NAME, Scope
@@ -29,26 +29,26 @@ def get_instance(cls, name: str=None) -> Self:
 
         return cls._instances[(cls, name)]
 
-    def register_transient(self, dependency: Type, implementation: Optional[Type] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
+    def register_transient(self, dependency: Type, implementation: Optional[Type] = None, tags: Optional[set] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
         if implementation is None:
             implementation = dependency
         if dependency in self._registrations:
             raise ValueError(f"Dependency {dependency} is already registered.")
-        self._registrations[dependency] = Registration(dependency, implementation, Scope.TRANSIENT, constructor_args)
+        self._registrations[dependency] = Registration(dependency, implementation, Scope.TRANSIENT, tags, constructor_args)
 
-    def register_scoped(self, dependency: Type, implementation: Optional[Type] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
+    def register_scoped(self, dependency: Type, implementation: Optional[Type] = None, tags: Optional[set] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
         if implementation is None:
             implementation = dependency
         if dependency in self._registrations:
             raise ValueError(f"Dependency {dependency} is already registered.")
-        self._registrations[dependency] = Registration(dependency, implementation, Scope.SCOPED, constructor_args)
+        self._registrations[dependency] = Registration(dependency, implementation, Scope.SCOPED, tags, constructor_args)
 
-    def register_singleton(self, dependency: Type, implementation: Optional[Type] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
+    def register_singleton(self, dependency: Type, implementation: Optional[Type] = None, tags: Optional[set] = None, constructor_args: Optional[Dict[str, Any]] = None) -> None:
         if implementation is None:
             implementation = dependency
         if dependency in self._registrations:
             raise ValueError(f"Dependency {dependency} is already registered.")
-        self._registrations[dependency] = Registration(dependency, implementation, Scope.SINGLETON, constructor_args)
+        self._registrations[dependency] = Registration(dependency, implementation, Scope.SINGLETON, tags, constructor_args)
 
     def resolve(self, dependency: Type, scope_name: str = DEFAULT_SCOPE_NAME) -> Type:
         if scope_name not in self._scoped_instances:
@@ -90,6 +90,15 @@ def resolve(self, dependency: Type, scope_name: str = DEFAULT_SCOPE_NAME) -> Typ
 
         raise ValueError(f"Invalid dependency scope: {scope}")
 
+    def resolve_all(self, tags: Optional[set] = None) -> List[Any]:
+        tags = [] if not tags else tags
+        resolved_dependencies = []
+        for registration in self._registrations.values():
+            if not len(tags) or tags.intersection(registration.tags):
+                resolved_dependencies.append(
+                    self.resolve(registration.dependency))
+        return resolved_dependencies
+
     def _validate_constructor_args(self, constructor_args: Dict[str, Any], implementation: Type) -> None:
         constructor = inspect.signature(implementation.__init__).parameters
 

src/dependency_injection/registration.py

@@ -5,8 +5,9 @@
 
 class Registration():
 
-    def __init__(self, dependency: Type, implementation: Type, scope: Scope, constructor_args: Optional[Dict[str, Any]] = None):
+    def __init__(self, dependency: Type, implementation: Type, scope: Scope, tags: Optional[set] = None, constructor_args: Optional[Dict[str, Any]] = None):
         self.dependency = dependency
         self.implementation = implementation
         self.scope = scope
+        self.tags = tags or set()
         self.constructor_args = constructor_args or {}

tests/unit_test/container/register/test_register_transient.py

@@ -44,7 +44,7 @@ class Car(Vehicle):
         with pytest.raises(ValueError, match="is already registered"):
             dependency_container.register_transient(dependency, implementation)
 
-    def test_register_transient_when_dependency_and_implementation_being_the_same(
+    def test_register_transient_success_when_dependency_and_implementation_same(
         self,
     ):
         # arrange
@@ -58,3 +58,38 @@ class Vehicle:
         dependency_container.register_transient(dependency)
 
         # assert (no exception thrown)
+
+    def test_register_transient_fails_when_already_registered_and_dependency_and_implementation_same(
+        self,
+    ):
+        # arrange
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, Car)
+
+        # act + assert
+        with pytest.raises(ValueError, match="is already registered"):
+            dependency_container.register_transient(Vehicle)
+
+    def test_register_transient_success_when_other_dependency_registered_of_implementation_ancestor_class(
+        self,
+    ):
+        # arrange
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, Car)
+
+        # act
+        dependency_container.register_transient(Car)
+
+        # assert (no exception thrown)

tests/unit_test/container/resolve/test_resolve_all.py

@@ -0,0 +1,123 @@
+from dependency_injection.container import DependencyContainer
+from unit_test.unit_test_case import UnitTestCase
+
+
+class TestResolveAll(UnitTestCase):
+
+    def test_returns_dependency_with_tag(
+        self,
+    ):
+        # arrange
+        class Driveable:
+            pass
+
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, Car, tags={Driveable})
+
+        # act
+        resolved_dependencies = dependency_container.resolve_all(tags={Driveable})
+
+        # assert
+        self.assertEqual(len(resolved_dependencies), 1)
+        self.assertIsInstance(resolved_dependencies[0], Car)
+
+    def test_returns_all_dependencies_with_tag(
+        self,
+    ):
+        # arrange
+        class Driveable:
+            pass
+
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        class Innovation:
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, tags={Driveable})
+        dependency_container.register_transient(Car, tags={Driveable})
+        dependency_container.register_transient(Innovation, tags={Driveable})
+
+        # act
+        resolved_dependencies = dependency_container.resolve_all(tags={Driveable})
+
+        # assert
+        self.assertEqual(len(resolved_dependencies), 3)
+        self.assertTrue(any(isinstance(dependency, Vehicle) for dependency in resolved_dependencies))
+        self.assertTrue(any(isinstance(dependency, Car) for dependency in resolved_dependencies))
+        self.assertTrue(any(isinstance(dependency, Innovation) for dependency in resolved_dependencies))
+
+    def test_does_not_return_dependency_without_tag(
+        self,
+    ):
+        # arrange
+        class Driveable:
+            pass
+
+        class Refuelable:
+            pass
+
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        class Innovation:
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, tags={Driveable, Refuelable})
+        dependency_container.register_transient(Car, tags={Driveable, Refuelable})
+        dependency_container.register_transient(Innovation, tags={Driveable})
+
+        # act
+        resolved_dependencies = dependency_container.resolve_all(tags={Refuelable})
+
+        # assert
+        self.assertEqual(len(resolved_dependencies), 2)
+        self.assertTrue(any(isinstance(dependency, Vehicle) for dependency in resolved_dependencies))
+        self.assertTrue(any(isinstance(dependency, Car) for dependency in resolved_dependencies))
+
+    def test_returns_all_dependencies_when_no_tag_specified(
+        self,
+    ):
+        # arrange
+        class Driveable:
+            pass
+
+        class Refuelable:
+            pass
+
+        class Vehicle:
+            pass
+
+        class Car(Vehicle):
+            pass
+
+        class Innovation:
+            pass
+
+        dependency_container = DependencyContainer.get_instance()
+        dependency_container.register_transient(Vehicle, tags={Driveable, Refuelable})
+        dependency_container.register_transient(Car, tags={Driveable, Refuelable})
+        dependency_container.register_transient(Innovation, tags={Driveable})
+
+        # act
+        resolved_dependencies = dependency_container.resolve_all()
+
+        # assert
+        self.assertEqual(len(resolved_dependencies), 3)
+        self.assertTrue(any(isinstance(dependency, Vehicle) for dependency in resolved_dependencies))
+        self.assertTrue(any(isinstance(dependency, Car) for dependency in resolved_dependencies))
+        self.assertTrue(any(isinstance(dependency, Innovation) for dependency in resolved_dependencies))