feat: ✨ Scoped dependencies

Author: remimd

README.md

@@ -56,6 +56,7 @@ if __name__ == "__main__":
 ## Resources
 
 * [**Basic usage**](https://github.com/100nm/python-injection/tree/prod/documentation/basic-usage.md)
+* [**Scoped dependencies**](https://github.com/100nm/python-injection/tree/prod/documentation/scoped-dependencies.md)
 * [**Testing**](https://github.com/100nm/python-injection/tree/prod/documentation/testing.md)
 * [**Advanced usage**](https://github.com/100nm/python-injection/tree/prod/documentation/advanced-usage.md)
 * [**Utils**](https://github.com/100nm/python-injection/tree/prod/documentation/utils.md)

documentation/scoped-dependencies.md

@@ -0,0 +1,98 @@
+# Scoped dependencies
+
+The scoped dependencies were created for two reasons:
+* To have dependencies that have a defined lifespan.
+* To be able to open and close things in a dependency recipe.
+
+## Best practices
+
+* Avoid making a singleton dependent on a scoped dependency.
+* Define scope names in a `StrEnum`.
+
+## Scope
+
+The scope is responsible for instance persistence and for cleaning up when it closes.
+
+There are two kinds of scopes:
+* **Contextual**: All threads have access to a different scope (based on [contextvars](https://docs.python.org/3.13/library/contextvars.html)).
+* **Shared**: All threads have access to the same scope.
+
+First of all, the scope must be defined:
+
+*By default, the `shared` parameter is `False`.*
+
+> Define an asynchronous scope:
+
+```python
+from injection import adefine_scope
+
+async def main() -> None:
+    async with adefine_scope("<scope-name>", shared=True):
+        ...
+```
+
+> Define a synchronous scope:
+
+```python
+from injection import define_scope
+
+def main() -> None:
+    with define_scope("<scope-name>", shared=True):
+        ...
+```
+
+## Register a scoped dependencies
+
+`@scoped` works exactly like `@injectable`, it just has extra features.
+
+### "contextmanager-like" recipes
+
+*Anything after the `yield` keyword will be executed when the scope is closed.*
+
+> Asynchronous (asynchronous scope required):
+
+```python
+from collections.abc import AsyncIterator
+from injection import scoped
+
+class Client:
+    async def open_connection(self) -> None: ...
+    
+    async def close_connection(self) -> None: ...
+
+@scoped("<scope-name>")
+async def client_recipe() -> AsyncIterator[Client]:
+    # On resolving dependency
+    client = Client()
+    await client.open_connection()
+    
+    try:
+        yield client
+    finally:
+        # On scope close
+        await client.close_connection()
+```
+
+> Synchronous:
+
+```python
+from collections.abc import Iterator
+from injection import scoped
+
+class Client:
+    def open_connection(self) -> None: ...
+    
+    def close_connection(self) -> None: ...
+
+@scoped("<scope-name>")
+def client_recipe() -> Iterator[Client]:
+    # On resolving dependency
+    client = Client()
+    client.open_connection()
+    
+    try:
+        yield client
+    finally:
+        # On scope close
+        client.close_connection()
+```

injection/__init__.py

@@ -1,23 +1,27 @@
 from ._core.descriptors import LazyInstance
 from ._core.injectables import Injectable
 from ._core.module import Mode, Module, Priority, mod
+from ._core.scope import adefine_scope, define_scope
 
 __all__ = (
     "Injectable",
     "LazyInstance",
     "Mode",
     "Module",
     "Priority",
+    "adefine_scope",
     "afind_instance",
     "aget_instance",
     "aget_lazy_instance",
     "constant",
+    "define_scope",
     "find_instance",
     "get_instance",
     "get_lazy_instance",
     "inject",
     "injectable",
     "mod",
+    "scoped",
     "set_constant",
     "should_be_injectable",
     "singleton",
@@ -32,6 +36,7 @@
 get_lazy_instance = mod().get_lazy_instance
 inject = mod().inject
 injectable = mod().injectable
+scoped = mod().scoped
 set_constant = mod().set_constant
 should_be_injectable = mod().should_be_injectable
 singleton = mod().singleton

injection/__init__.pyi

@@ -1,39 +1,36 @@
 from abc import abstractmethod
-from collections.abc import Awaitable, Callable
-from contextlib import ContextDecorator
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from contextlib import asynccontextmanager, contextmanager
 from enum import Enum
 from logging import Logger
-from typing import (
-    Any,
-    ContextManager,
-    Protocol,
-    Self,
-    final,
-    overload,
-    runtime_checkable,
-)
+from typing import Any, Final, Protocol, Self, final, overload, runtime_checkable
 
 from ._core.common.invertible import Invertible as _Invertible
 from ._core.common.type import InputType as _InputType
 from ._core.common.type import TypeInfo as _TypeInfo
 from ._core.module import InjectableFactory as _InjectableFactory
 from ._core.module import ModeStr, PriorityStr
 
-__module: Module = ...
+__MODULE: Final[Module] = ...
 
-afind_instance = __module.afind_instance
-aget_instance = __module.aget_instance
-aget_lazy_instance = __module.aget_lazy_instance
-constant = __module.constant
-find_instance = __module.find_instance
-get_instance = __module.get_instance
-get_lazy_instance = __module.get_lazy_instance
-inject = __module.inject
-injectable = __module.injectable
-set_constant = __module.set_constant
-should_be_injectable = __module.should_be_injectable
-singleton = __module.singleton
+afind_instance = __MODULE.afind_instance
+aget_instance = __MODULE.aget_instance
+aget_lazy_instance = __MODULE.aget_lazy_instance
+constant = __MODULE.constant
+find_instance = __MODULE.find_instance
+get_instance = __MODULE.get_instance
+get_lazy_instance = __MODULE.get_lazy_instance
+inject = __MODULE.inject
+injectable = __MODULE.injectable
+scoped = __MODULE.scoped
+set_constant = __MODULE.set_constant
+should_be_injectable = __MODULE.should_be_injectable
+singleton = __MODULE.singleton
 
+@asynccontextmanager
+def adefine_scope(name: str, *, shared: bool = ...) -> AsyncIterator[None]: ...
+@contextmanager
+def define_scope(name: str, *, shared: bool = ...) -> Iterator[None]: ...
 def mod(name: str = ..., /) -> Module:
     """
     Short syntax for `Module.from_name`.
@@ -109,6 +106,21 @@ class Module:
         always be the same.
         """
 
+    def scoped[**P, T](
+        self,
+        scope_name: str,
+        /,
+        *,
+        inject: bool = ...,
+        on: _TypeInfo[T] = (),
+        mode: Mode | ModeStr = ...,
+    ) -> Any:
+        """
+        Decorator applicable to a class or function or generator function. It is used
+        to indicate how the scoped instance will be constructed. At injection time, the
+        injected instance is retrieved from the scope.
+        """
+
     def should_be_injectable[T](self, wrapped: type[T] = ..., /) -> Any:
         """
         Decorator applicable to a class. It is used to specify whether an injectable
@@ -248,12 +260,13 @@ class Module:
         Function to remove a module in use.
         """
 
+    @contextmanager
     def use_temporarily(
         self,
         module: Module,
         *,
         priority: Priority | PriorityStr = ...,
-    ) -> ContextManager[None] | ContextDecorator:
+    ) -> Iterator[None]:
         """
         Context manager or decorator for temporary use of a module.
         """

injection/_core/common/asynchronous.py

@@ -1,10 +1,19 @@
 import asyncio
 from abc import abstractmethod
-from collections.abc import Awaitable, Callable, Generator
+from collections.abc import Awaitable, Callable, Coroutine, Generator
 from dataclasses import dataclass
 from typing import Any, Protocol, runtime_checkable
 
 
+def run_sync[T](coroutine: Coroutine[Any, Any, T]) -> T:
+    loop = asyncio.get_event_loop()
+
+    try:
+        return loop.run_until_complete(coroutine)
+    finally:
+        coroutine.close()
+
+
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
 class SimpleAwaitable[T](Awaitable[T]):
     callable: Callable[..., Awaitable[T]]
@@ -28,21 +37,13 @@ def call(self, /, *args: P.args, **kwargs: P.kwargs) -> T:
 
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
 class AsyncCaller[**P, T](Caller[P, T]):
-    callable: Callable[P, Awaitable[T]]
+    callable: Callable[P, Coroutine[Any, Any, T]]
 
     async def acall(self, /, *args: P.args, **kwargs: P.kwargs) -> T:
         return await self.callable(*args, **kwargs)
 
     def call(self, /, *args: P.args, **kwargs: P.kwargs) -> T:
-        loop = asyncio.get_event_loop()
-
-        if loop.is_running():
-            raise RuntimeError(
-                "Can't call an asynchronous function in a synchronous context."
-            )
-
-        coroutine = self.callable(*args, **kwargs)
-        return loop.run_until_complete(coroutine)
+        return run_sync(self.callable(*args, **kwargs))
 
 
 @dataclass(repr=False, eq=False, frozen=True, slots=True)

injection/_core/common/key.py

@@ -0,0 +1,5 @@
+from uuid import uuid4
+
+
+def new_short_key() -> str:
+    return uuid4().hex[:7]

injection/_core/common/type.py

@@ -1,4 +1,12 @@
-from collections.abc import Callable, Iterable, Iterator
+from collections.abc import (
+    AsyncGenerator,
+    AsyncIterable,
+    AsyncIterator,
+    Callable,
+    Generator,
+    Iterable,
+    Iterator,
+)
 from inspect import isfunction
 from types import GenericAlias, UnionType
 from typing import (
@@ -23,7 +31,7 @@ def get_return_types(*args: TypeInfo[Any]) -> Iterator[InputType[Any]]:
         ):
             inner_args = arg
 
-        elif isfunction(arg) and (return_type := get_type_hints(arg).get("return")):
+        elif isfunction(arg) and (return_type := get_return_hint(arg)):
             inner_args = (return_type,)
 
         else:
@@ -33,6 +41,29 @@ def get_return_types(*args: TypeInfo[Any]) -> Iterator[InputType[Any]]:
         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](
+    function: Callable[..., Iterator[T]] | Callable[..., AsyncIterator[T]],
+) -> InputType[T] | None:
+    return_type = get_return_hint(function)
+
+    if get_origin(return_type) not in {
+        AsyncGenerator,
+        AsyncIterable,
+        AsyncIterator,
+        Generator,
+        Iterable,
+        Iterator,
+    }:
+        return None
+
+    args = get_args(return_type)
+    return next(iter(args), None)
+
+
 def standardize_types(
     *types: InputType[Any],
     with_origin: bool = False,