refactor: @scoped

Author: remimd

injection/_core/common/type.py

@@ -28,39 +28,39 @@
 )
 
 
-def get_return_types(*args: TypeInfo[Any]) -> Iterator[InputType[Any]]:
-    for arg in args:
-        if isinstance(arg, Collection) and not isclass(arg):
-            inner_args = arg
-
-        elif isfunction(arg) and (return_type := get_return_hint(arg)):
-            inner_args = (return_type,)
-
-        else:
-            yield arg  # type: ignore[misc]
-            continue
-
-        yield from get_return_types(*inner_args)
-
-
 def get_return_hint[T](function: Callable[..., T]) -> InputType[T] | None:
     return get_type_hints(function).get("return")
 
 
-def get_yield_hint[T](
+def get_yield_hints[T](
     function: Callable[..., Iterator[T]] | Callable[..., AsyncIterator[T]],
 ) -> tuple[InputType[T]] | tuple[()]:
     return_type = get_return_hint(function)
 
-    if get_origin(return_type) in {
+    if get_origin(return_type) in (
         AsyncGenerator,
         AsyncIterable,
         AsyncIterator,
         Generator,
         Iterable,
         Iterator,
-    }:
+    ):
         for arg in get_args(return_type):
             return (arg,)
 
     return ()
+
+
+def iter_return_types(*args: TypeInfo[Any]) -> Iterator[InputType[Any]]:
+    for arg in args:
+        if isinstance(arg, Collection) and not isclass(arg):
+            inner_args = arg
+
+        elif isfunction(arg) and (return_type := get_return_hint(arg)):
+            inner_args = (return_type,)
+
+        else:
+            yield arg  # type: ignore[misc]
+            continue
+
+        yield from iter_return_types(*inner_args)

injection/_core/injectables.py

@@ -10,6 +10,7 @@
     ContextManager,
     NoReturn,
     Protocol,
+    Self,
     runtime_checkable,
 )
 
@@ -166,6 +167,10 @@ def unlock(self) -> None:
     def __get_scope(self) -> Scope:
         return get_scope(self.scope_name)
 
+    @classmethod
+    def bind_scope_name(cls, name: str) -> Callable[[Caller[..., R]], Self]:
+        return partial(cls, scope_name=name)
+
 
 class AsyncCMScopedInjectable[T](ScopedInjectable[AsyncContextManager[T], T]):
     __slots__ = ()

injection/_core/module.py

@@ -17,7 +17,7 @@
 from contextlib import asynccontextmanager, contextmanager, suppress
 from dataclasses import dataclass, field
 from enum import StrEnum
-from functools import partial, partialmethod, singledispatchmethod, update_wrapper
+from functools import partialmethod, singledispatchmethod, update_wrapper
 from inspect import (
     BoundArguments,
     Signature,
@@ -57,8 +57,8 @@
 from injection._core.common.type import (
     InputType,
     TypeInfo,
-    get_return_types,
-    get_yield_hint,
+    get_yield_hints,
+    iter_return_types,
 )
 from injection._core.injectables import (
     AsyncCMScopedInjectable,
@@ -169,14 +169,21 @@ def get_default(cls) -> Priority:
 
 type PriorityStr = Literal["low", "high"]
 
-type ContextManagerLikeRecipe[**P, T] = (
+type ContextManagerRecipe[**P, T] = (
     Callable[P, ContextManager[T]] | Callable[P, AsyncContextManager[T]]
 )
 type GeneratorRecipe[**P, T] = (
     Callable[P, Generator[T, Any, Any]] | Callable[P, AsyncGenerator[T, Any]]
 )
 
 
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
+class _ScopedContext[**P, T]:
+    cls: type[ScopedInjectable[Any, T]]
+    hints: TypeInfo[T]
+    wrapper: Recipe[P, T] | ContextManagerRecipe[P, T]
+
+
 @dataclass(eq=False, frozen=True, slots=True)
 class Module(EventListener, InjectionProvider):  # type: ignore[misc]
     name: str = field(default_factory=lambda: f"anonymous@{new_short_key()}")
@@ -266,29 +273,33 @@ def scoped[**P, T](
         def decorator(
             wrapped: Recipe[P, T] | GeneratorRecipe[P, T],
         ) -> Recipe[P, T] | GeneratorRecipe[P, T]:
-            injectable_class: type[ScopedInjectable[Any, T]]
-            wrapper: Recipe[P, T] | ContextManagerLikeRecipe[P, T]
-
             if isasyncgenfunction(wrapped):
-                hint = get_yield_hint(wrapped)
-                injectable_class = AsyncCMScopedInjectable
-                wrapper = asynccontextmanager(wrapped)
+                ctx = _ScopedContext(
+                    cls=AsyncCMScopedInjectable,
+                    hints=get_yield_hints(wrapped),
+                    wrapper=asynccontextmanager(wrapped),
+                )
 
             elif isgeneratorfunction(wrapped):
-                hint = get_yield_hint(wrapped)
-                injectable_class = CMScopedInjectable
-                wrapper = contextmanager(wrapped)
+                ctx = _ScopedContext(
+                    cls=CMScopedInjectable,
+                    hints=get_yield_hints(wrapped),
+                    wrapper=contextmanager(wrapped),
+                )
 
             else:
-                injectable_class = SimpleScopedInjectable
-                hint = wrapper = wrapped  # type: ignore[assignment]
+                ctx = _ScopedContext(
+                    cls=SimpleScopedInjectable,
+                    hints=(wrapped,),
+                    wrapper=wrapped,
+                )
 
             self.injectable(
-                wrapper,
-                cls=partial(injectable_class, scope_name=scope_name),
+                ctx.wrapper,
+                cls=ctx.cls.bind_scope_name(scope_name),
                 ignore_type_hint=True,
                 inject=inject,
-                on=(hint, on),
+                on=(ctx.hints, on),
                 mode=mode,
             )
             return wrapped
@@ -715,7 +726,7 @@ def __build_key_types(input_cls: Any) -> frozenset[Any]:
         config = MatchingTypesConfig(ignore_none=True)
         return frozenset(
             itertools.chain.from_iterable(
-                iter_matching_types(cls, config) for cls in get_return_types(input_cls)
+                iter_matching_types(cls, config) for cls in iter_return_types(input_cls)
             )
         )