Check scoped_services before resolving from map when in a scope

Author: StummeJ

rodi/__init__.py

@@ -79,7 +79,7 @@ def _get_obj_locals(obj) -> Optional[Dict[str, Any]]:
 
 
 def class_name(input_type):
-    if input_type in {list, set} and str(
+    if input_type in {list, set} and str(  # noqa: E721
         type(input_type) == "<class 'types.GenericAlias'>"
     ):
         # for Python 3.9 list[T], set[T]
@@ -233,9 +233,9 @@ class ActivationScope:
     def __init__(
         self,
         provider: Optional["Services"] = None,
-        scoped_services: Optional[Dict[Type[T], T]] = None,
+        scoped_services: Optional[Dict[Union[Type[T], str], T]] = None,
     ):
-        self.provider = provider
+        self.provider = provider or Services()
         self.scoped_services = scoped_services or {}
 
     def __enter__(self):
@@ -248,7 +248,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def get(
         self,
-        desired_type: Union[Type[T], str],
+        desired_type: Union[Union[Type[T], str], str],
         scope: Optional["ActivationScope"] = None,
         *,
         default: Optional[Any] = ...,
@@ -713,13 +713,14 @@ def get(
             scope = ActivationScope(self)
 
         resolver = self._map.get(desired_type)
+        scoped_service = scope.scoped_services.get(desired_type) if scope else None
 
-        if not resolver:
+        if not resolver and not scoped_service:
             if default is not ...:
                 return cast(T, default)
             raise CannotResolveTypeException(desired_type)
 
-        return cast(T, resolver(scope, desired_type))
+        return cast(T, scoped_service or resolver(scope, desired_type))
 
     def _get_getter(self, key, param):
         if param.annotation is _empty:
@@ -756,13 +757,15 @@ def get_executor(self, method: Callable) -> Callable:
 
         if iscoroutinefunction(method):
 
-            async def async_executor(scoped: Optional[Dict[Type, Any]] = None):
+            async def async_executor(
+                scoped: Optional[Dict[Union[Type, str], Any]] = None
+            ):
                 with ActivationScope(self, scoped) as context:
                     return await method(*[fn(context) for fn in fns])
 
             return async_executor
 
-        def executor(scoped: Optional[Dict[Type, Any]] = None):
+        def executor(scoped: Optional[Dict[Union[Type, str], Any]] = None):
             with ActivationScope(self, scoped) as context:
                 return method(*[fn(context) for fn in fns])
 

tests/test_services.py

@@ -505,6 +505,18 @@ def test_scoped_services_use_correct_scope_context_by_default_with_multiple_scop
     assert c is not f
 
 
+def test_scoped_services_works_with_str_keys():
+    container = Container()
+    container.add_singleton("Id", IdGetter)
+    provider = container.build_provider()
+
+    with ActivationScope(provider) as scoped_provider:
+        a = scoped_provider.get("Id")
+    b = provider.get("id")
+
+    assert a is b
+
+
 def test_scoped_services():
     container = Container()
     container._add_exact_scoped(IdGetter)
@@ -522,6 +534,37 @@ def test_scoped_services():
     assert b is not d
 
 
+def test_scoped_service_from_scoped_services():
+    container = Container()
+    provider = container.build_provider()
+
+    scoped_service = IdGetter()
+
+    with ActivationScope(
+        provider,
+        {
+            IdGetter: scoped_service,
+        },
+    ) as context:
+        a = provider.get(IdGetter, context)
+        b = provider.get(IdGetter, default=None)
+    c = provider.get(IdGetter, default=None)
+
+    with ActivationScope(
+        scoped_services={
+            IdGetter: scoped_service,
+        }
+    ) as context:
+        d = provider.get(IdGetter, context)
+        e = provider.get(IdGetter, default=None)
+
+    assert a is scoped_service
+    assert b is None
+    assert c is None
+    assert d is scoped_service
+    assert e is None
+
+
 def test_scoped_services_with_shortcut():
     container = Container()
     container.add_scoped(IdGetter)