refactor: Internal logic about constants

Author: remimd

injection/_core/common/lazy.py

@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from collections.abc import Callable, Iterator
 from functools import partial
 
 from injection._core.common.invertible import Invertible
@@ -13,15 +13,6 @@ def cache() -> Iterator[T]:
     return partial(next, cache())
 
 
-def alazy[T](factory: Callable[..., Awaitable[T]]) -> Callable[[], Awaitable[T]]:
-    async def cache() -> AsyncIterator[T]:
-        value = await factory()
-        while True:
-            yield value
-
-    return partial(_anext, cache())
-
-
 class Lazy[T](Invertible[T]):
     __slots__ = ("__get", "__is_set")
 
@@ -44,7 +35,3 @@ def __invert__(self) -> T:
     @property
     def is_set(self) -> bool:
         return self.__is_set
-
-
-async def _anext[T](async_iterator: AsyncIterator[T]) -> T:
-    return await anext(async_iterator)

injection/_core/injectables.py

@@ -119,6 +119,17 @@ def unlock(self) -> None:
         self.cache.pop(self.__key, None)
 
 
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
+class ConstantInjectable[T](Injectable[T]):
+    instance: T
+
+    async def aget_instance(self) -> T:
+        return self.instance
+
+    def get_instance(self) -> T:
+        return self.instance
+
+
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
 class ScopedInjectable[R, T](Injectable[T], ABC):
     factory: Caller[..., R]

injection/_core/locator.py

@@ -73,32 +73,44 @@ class InjectableBroker[T](Protocol):
     def get(self, provider: InjectionProvider) -> Injectable[T] | None:
         raise NotImplementedError
 
+    @abstractmethod
+    def is_locked(self, provider: InjectionProvider) -> bool:
+        raise NotImplementedError
+
     @abstractmethod
     def request(self, provider: InjectionProvider) -> Injectable[T]:
         raise NotImplementedError
 
 
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
 class DynamicInjectableBroker[T](InjectableBroker[T]):
-    injectable_factory: InjectableFactory[T]
+    factory: InjectableFactory[T]
     recipe: Recipe[..., T]
-    cache: WeakKeyDictionary[InjectionProvider, Injectable[T]] = field(
+    injectables: WeakKeyDictionary[InjectionProvider, Injectable[T]] = field(
         default_factory=WeakKeyDictionary,
         init=False,
     )
 
     def get(self, provider: InjectionProvider) -> Injectable[T] | None:
-        return self.cache.get(provider)
+        return self.injectables.get(provider)
+
+    def is_locked(self, provider: InjectionProvider) -> bool:
+        injectable = self.get(provider)
+
+        if injectable is None:
+            return False
+
+        return injectable.is_locked
 
     def request(self, provider: InjectionProvider) -> Injectable[T]:
         with suppress(KeyError):
-            return self.cache[provider]
+            return self.injectables[provider]
 
         injectable = _make_injectable(
-            self.injectable_factory,
+            self.factory,
             provider.make_injected_function(self.recipe),  # type: ignore[misc]
         )
-        self.cache[provider] = injectable
+        self.injectables[provider] = injectable
         return injectable
 
 
@@ -109,16 +121,19 @@ class StaticInjectableBroker[T](InjectableBroker[T]):
     def get(self, provider: InjectionProvider) -> Injectable[T] | None:
         return self.value
 
+    def is_locked(self, provider: InjectionProvider) -> bool:
+        return False
+
     def request(self, provider: InjectionProvider) -> Injectable[T]:
         return self.value
 
     @classmethod
     def from_factory(
         cls,
-        injectable_factory: InjectableFactory[T],
+        factory: InjectableFactory[T],
         recipe: Recipe[..., T],
     ) -> Self:
-        return cls(_make_injectable(injectable_factory, recipe))
+        return cls(_make_injectable(factory, recipe))
 
 
 class Mode(StrEnum):
@@ -172,9 +187,7 @@ def __brokers(self) -> frozenset[InjectableBroker[Any]]:
         return frozenset(record.broker for record in self.__records.values())
 
     def is_locked(self, provider: InjectionProvider) -> bool:
-        return any(
-            injectable.is_locked for injectable in self.__iter_injectables(provider)
-        )
+        return any(broker.is_locked(provider) for broker in self.__brokers)
 
     def request[T](
         self,

injection/_core/module.py

@@ -50,7 +50,7 @@
 from injection._core.common.event import Event, EventChannel, EventListener
 from injection._core.common.invertible import Invertible, SimpleInvertible
 from injection._core.common.key import new_short_key
-from injection._core.common.lazy import Lazy, alazy, lazy
+from injection._core.common.lazy import Lazy
 from injection._core.common.threading import get_lock
 from injection._core.common.type import (
     InputType,
@@ -61,6 +61,7 @@
 from injection._core.injectables import (
     AsyncCMScopedInjectable,
     CMScopedInjectable,
+    ConstantInjectable,
     Injectable,
     ScopedInjectable,
     ScopedSlotInjectable,
@@ -249,6 +250,7 @@ def decorator(wp: Recipe[P, T]) -> Recipe[P, T]:
         return decorator(wrapped) if wrapped else decorator
 
     singleton = partialmethod(injectable, cls=SingletonInjectable)
+    constant = partialmethod(injectable, cls=SingletonInjectable, inject=False)
 
     def scoped[**P, T](
         self,
@@ -293,34 +295,12 @@ def decorator(
 
     def should_be_injectable[T](self, wrapped: type[T] | None = None, /) -> Any:
         def decorator(wp: type[T]) -> type[T]:
-            injectable = ShouldBeInjectable(wp)
-            broker = StaticInjectableBroker(injectable)
+            broker = StaticInjectableBroker(ShouldBeInjectable(wp))
             self.update_from(wp, broker, Mode.FALLBACK)
             return wp
 
         return decorator(wrapped) if wrapped else decorator
 
-    def constant[**P, T](
-        self,
-        wrapped: Recipe[P, T] | None = None,
-        /,
-        *,
-        on: TypeInfo[T] = (),
-        mode: Mode | ModeStr = Mode.get_default(),
-    ) -> Any:
-        def decorator(wp: Recipe[P, T]) -> Recipe[P, T]:
-            recipe: Recipe[[], T] = alazy(wp) if iscoroutinefunction(wp) else lazy(wp)  # type: ignore[arg-type]
-            self.injectable(
-                recipe,
-                ignore_type_hint=True,
-                inject=False,
-                on=(wp, on),
-                mode=mode,
-            )
-            return wp
-
-        return decorator(wrapped) if wrapped else decorator
-
     def set_constant[T](
         self,
         instance: T,
@@ -335,13 +315,8 @@ def set_constant[T](
         elif not on:
             raise ValueError("`on` must be provided when `alias` is `True`.")
 
-        self.injectable(
-            lambda: instance,
-            ignore_type_hint=True,
-            inject=False,
-            on=on,
-            mode=mode,
-        )
+        broker = StaticInjectableBroker(ConstantInjectable(instance))
+        self.update_from(on, broker, mode)
         return instance
 
     def reserve_scoped_slot[T](