feat: ⚡️ Scoped slot rework

Author: remimd

documentation/integrations.md

@@ -58,7 +58,7 @@ class InjectionScope(StrEnum):
 
 @asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncIterator[None]:
-    async with adefine_scope(InjectionScope.LIFESPAN, shared=True):
+    async with adefine_scope(InjectionScope.LIFESPAN, kind="shared"):
         yield
 
 app = FastAPI(lifespan=lifespan)
@@ -70,7 +70,7 @@ async def define_request_scope_middleware(
     request: Request,
     handler: Callable[[Request], Awaitable[Response]],
 ) -> Response:
-    async with adefine_scope(InjectionScope.REQUEST):
-        request_slot.set(request)
+    async with adefine_scope(InjectionScope.REQUEST) as scope:
+        scope.set_slot(request_slot, request)
         return await handler(request)
 ```

documentation/scoped-dependencies.md

@@ -19,15 +19,15 @@ There are two kinds of scopes:
 
 First of all, the scope must be defined:
 
-_By default, the `shared` parameter is `False`._
+_By default, the `kind` parameter is `"contextual"`._
 
 > Define an asynchronous scope:
 
 ```python
 from injection import adefine_scope
 
 async def main() -> None:
-    async with adefine_scope("<scope-name>", shared=True):
+    async with adefine_scope("<scope-name>"):
         ...
 ```
 
@@ -37,7 +37,7 @@ async def main() -> None:
 from injection import define_scope
 
 def main() -> None:
-    with define_scope("<scope-name>", shared=True):
+    with define_scope("<scope-name>"):
         ...
 ```
 
@@ -113,7 +113,7 @@ class Request: ...
 request_slot = reserve_scoped_slot(Request, scope_name="request")
 
 def process_request(request: Request) -> None:
-    with define_scope("request"):
-        request_slot.set(request)
+    with define_scope("request") as scope:
+        scope.set_slot(request_slot, request)
         # ...
 ```

injection/__init__.py

@@ -1,7 +1,8 @@
 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
+from ._core.scope import ScopeFacade as Scope
+from ._core.scope import ScopeKind, adefine_scope, define_scope
 from ._core.slots import Slot
 
 __all__ = (
@@ -10,6 +11,8 @@
     "Mode",
     "Module",
     "Priority",
+    "Scope",
+    "ScopeKind",
     "Slot",
     "adefine_scope",
     "afind_instance",

injection/__init__.pyi

@@ -1,5 +1,5 @@
 from abc import abstractmethod
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping
 from contextlib import asynccontextmanager, contextmanager
 from enum import Enum
 from logging import Logger
@@ -11,6 +11,7 @@ from ._core.common.type import TypeInfo as _TypeInfo
 from ._core.module import InjectableFactory as _InjectableFactory
 from ._core.module import ModeStr, PriorityStr
 from ._core.module import Recipe as _Recipe
+from ._core.scope import ScopeKindStr
 
 __MODULE: Final[Module] = ...
 
@@ -30,9 +31,15 @@ should_be_injectable = __MODULE.should_be_injectable
 singleton = __MODULE.singleton
 
 @asynccontextmanager
-def adefine_scope(name: str, *, shared: bool = ...) -> AsyncIterator[None]: ...
+def adefine_scope(
+    name: str,
+    kind: ScopeKind | ScopeKindStr = ...,
+) -> AsyncIterator[Scope]: ...
 @contextmanager
-def define_scope(name: str, *, shared: bool = ...) -> Iterator[None]: ...
+def define_scope(
+    name: str,
+    kind: ScopeKind | ScopeKindStr = ...,
+) -> Iterator[Scope]: ...
 def mod(name: str = ..., /) -> Module:
     """
     Short syntax for `Module.from_name`.
@@ -47,10 +54,16 @@ class Injectable[T](Protocol):
     @abstractmethod
     def get_instance(self) -> T: ...
 
-@runtime_checkable
-class Slot[T](Protocol):
-    @abstractmethod
-    def set(self, instance: T, /) -> None: ...
+@final
+class ScopeKind(Enum):
+    CONTEXTUAL = ...
+    SHARED = ...
+
+class Scope:
+    def set_slot[T](self, slot: Slot[T], value: T) -> Self: ...
+    def slot_map(self, values: Mapping[Slot[Any], Any]) -> Self: ...
+
+class Slot[T]: ...
 
 class LazyInstance[T]:
     def __init__(
@@ -179,7 +192,7 @@ class Module:
 
     def reserve_scoped_slot[T](
         self,
-        on: _TypeInfo[T],
+        cls: type[T],
         /,
         scope_name: str,
         *,

injection/_core/injectables.py

@@ -17,8 +17,14 @@
 from injection._core.common.asynchronous import (
     create_semaphore as _create_async_semaphore,
 )
-from injection._core.scope import Scope, get_active_scopes, get_scope
-from injection.exceptions import InjectionError
+from injection._core.scope import (
+    Scope,
+    get_scope,
+    in_scope_cache,
+    remove_scoped_values,
+)
+from injection._core.slots import Slot
+from injection.exceptions import EmptySlotError, InjectionError
 
 
 @runtime_checkable
@@ -123,7 +129,7 @@ class ScopedInjectable[R, T](Injectable[T], ABC):
 
     @property
     def is_locked(self) -> bool:
-        return any(self in scope.cache for scope in get_active_scopes(self.scope_name))
+        return in_scope_cache(self, self.scope_name)
 
     @abstractmethod
     async def abuild(self, scope: Scope) -> T:
@@ -143,10 +149,6 @@ def get_instance(self) -> T:
         factory = partial(self.build, scope)
         return self.logic.get_or_create(scope.cache, self, factory)
 
-    def setdefault(self, instance: T) -> T:
-        scope = self.__get_scope()
-        return self.logic.get_or_create(scope.cache, self, lambda: instance)
-
     def unlock(self) -> None:
         if self.is_locked:
             raise RuntimeError(f"To unlock, close the `{self.scope_name}` scope.")
@@ -188,8 +190,35 @@ def build(self, scope: Scope) -> T:
         return self.factory.call()
 
     def unlock(self) -> None:
-        for scope in get_active_scopes(self.scope_name):
-            scope.cache.pop(self, None)
+        remove_scoped_values(self, self.scope_name)
+
+
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
+class ScopedSlotInjectable[T](Injectable[T]):
+    cls: type[T]
+    scope_name: str
+    slot: Slot[T] = field(default_factory=Slot)
+
+    @property
+    def is_locked(self) -> bool:
+        return in_scope_cache(self.slot, self.scope_name)
+
+    async def aget_instance(self) -> T:
+        return self.get_instance()
+
+    def get_instance(self) -> T:
+        scope_name = self.scope_name
+        scope = get_scope(scope_name)
+
+        try:
+            return scope.cache[self.slot]
+        except KeyError as exc:
+            raise EmptySlotError(
+                f"The slot for `{self.cls}` isn't set in the current `{scope_name}` scope."
+            ) from exc
+
+    def unlock(self) -> None:
+        remove_scoped_values(self.slot, self.scope_name)
 
 
 @dataclass(repr=False, eq=False, frozen=True, slots=True)

injection/_core/module.py

@@ -64,14 +64,14 @@
     CMScopedInjectable,
     Injectable,
     ScopedInjectable,
+    ScopedSlotInjectable,
     ShouldBeInjectable,
     SimpleInjectable,
     SimpleScopedInjectable,
     SingletonInjectable,
 )
-from injection._core.slots import ScopedSlot, Slot
+from injection._core.slots import Slot
 from injection.exceptions import (
-    EmptySlotError,
     ModuleError,
     ModuleLockError,
     ModuleNotUsedError,
@@ -543,21 +543,16 @@ def set_constant[T](
 
     def reserve_scoped_slot[T](
         self,
-        on: TypeInfo[T],
+        cls: type[T],
         /,
         scope_name: str,
         *,
         mode: Mode | ModeStr = Mode.get_default(),
     ) -> Slot[T]:
-        def when_empty() -> T:
-            raise EmptySlotError(
-                f"The slot for `{on}` isn't set in the current `{scope_name}` scope."
-            )
-
-        injectable = SimpleScopedInjectable(SyncCaller(when_empty), scope_name)
-        updater = Updater.with_basics(on, injectable, mode)
+        injectable = ScopedSlotInjectable(cls, scope_name)
+        updater = Updater.with_basics(cls, injectable, mode)
         self.update(updater)
-        return ScopedSlot(injectable)
+        return injectable.slot
 
     def inject[**P, T](
         self,

injection/_core/scope.py

@@ -7,12 +7,14 @@
 from contextlib import AsyncExitStack, ExitStack, asynccontextmanager, contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+from enum import StrEnum
 from types import EllipsisType, TracebackType
 from typing import (
     Any,
     AsyncContextManager,
     ContextManager,
     Final,
+    Literal,
     NoReturn,
     Protocol,
     Self,
@@ -21,13 +23,27 @@
 )
 
 from injection._core.common.key import new_short_key
+from injection._core.slots import Slot
 from injection.exceptions import (
+    InjectionError,
     ScopeAlreadyDefinedError,
     ScopeError,
     ScopeUndefinedError,
 )
 
 
+class ScopeKind(StrEnum):
+    CONTEXTUAL = "contextual"
+    SHARED = "shared"
+
+    @classmethod
+    def get_default(cls) -> ScopeKind:
+        return cls.CONTEXTUAL
+
+
+type ScopeKindStr = Literal["contextual", "shared"]
+
+
 @runtime_checkable
 class ScopeState(Protocol):
     __slots__ = ()
@@ -109,17 +125,21 @@ def get_scope(self) -> Scope | None:
 
 
 @asynccontextmanager
-async def adefine_scope(name: str, *, shared: bool = False) -> AsyncIterator[None]:
+async def adefine_scope(
+    name: str,
+    kind: ScopeKind | ScopeKindStr = ScopeKind.get_default(),
+) -> AsyncIterator[ScopeFacade]:
     async with AsyncScope() as scope:
-        scope.enter(_bind_scope(name, scope, shared))
-        yield
+        yield scope.enter(_bind_scope(name, scope, kind))
 
 
 @contextmanager
-def define_scope(name: str, *, shared: bool = False) -> Iterator[None]:
+def define_scope(
+    name: str,
+    kind: ScopeKind | ScopeKindStr = ScopeKind.get_default(),
+) -> Iterator[ScopeFacade]:
     with SyncScope() as scope:
-        scope.enter(_bind_scope(name, scope, shared))
-        yield
+        yield scope.enter(_bind_scope(name, scope, kind))
 
 
 def get_active_scopes(name: str) -> tuple[Scope, ...]:
@@ -153,23 +173,40 @@ def get_scope(name, default=...):  # type: ignore[no-untyped-def]
     return default
 
 
-@contextmanager
-def _bind_scope(name: str, scope: Scope, shared: bool) -> Iterator[None]:
-    if shared:
-        is_already_defined = bool(get_active_scopes(name))
-        states = __SHARED_SCOPES
+def in_scope_cache(key: Any, scope_name: str) -> bool:
+    return any(key in scope.cache for scope in get_active_scopes(scope_name))
+
+
+def remove_scoped_values(key: Any, scope_name: str) -> None:
+    for scope in get_active_scopes(scope_name):
+        scope.cache.pop(key, None)
+
 
-    else:
-        is_already_defined = bool(get_scope(name, default=None))
-        states = __CONTEXTUAL_SCOPES
+@contextmanager
+def _bind_scope(
+    name: str,
+    scope: Scope,
+    kind: ScopeKind | ScopeKindStr,
+) -> Iterator[ScopeFacade]:
+    match ScopeKind(kind):
+        case ScopeKind.CONTEXTUAL:
+            is_already_defined = bool(get_scope(name, default=None))
+            states = __CONTEXTUAL_SCOPES
+
+        case ScopeKind.SHARED:
+            is_already_defined = bool(get_active_scopes(name))
+            states = __SHARED_SCOPES
+
+        case _:
+            raise NotImplementedError
 
     if is_already_defined:
         raise ScopeAlreadyDefinedError(
             f"Scope `{name}` is already defined in the current context."
         )
 
     with states[name].bind(scope):
-        yield
+        yield ScopeFacade(scope)
 
 
 @runtime_checkable
@@ -245,3 +282,21 @@ async def aenter[T](self, context_manager: AsyncContextManager[T]) -> NoReturn:
 
     def enter[T](self, context_manager: ContextManager[T]) -> T:
         return self.delegate.enter_context(context_manager)
+
+
+@dataclass(repr=False, frozen=True, slots=True)
+class ScopeFacade:
+    scope: Scope
+
+    def set_slot[T](self, slot: Slot[T], value: T) -> Self:
+        return self.slot_map({slot: value})
+
+    def slot_map(self, values: Mapping[Slot[Any], Any]) -> Self:
+        cache = self.scope.cache
+
+        for slot in values:
+            if slot in cache:
+                raise InjectionError("Slot already set.")
+
+        cache.update(values)
+        return self