feat: ✨ Inject self

Author: remimd

injection/_pkg.pyi

@@ -39,19 +39,11 @@ class Module:
 
     def __init__(self, name: str = ...): ...
     def __contains__(self, cls: type | UnionType, /) -> bool: ...
-    def inject(
-        self,
-        wrapped: Callable[..., Any] = ...,
-        /,
-        *,
-        force: bool = ...,
-    ):
+    def inject(self, wrapped: Callable[..., Any] = ..., /):
         """
         Decorator applicable to a class or function. Inject function dependencies using
         parameter type annotations. If applied to a class, the dependencies resolved
         will be those of the `__init__` method.
-
-        With `force=True`, parameters passed to replace dependencies will be ignored.
         """
 
     def injectable(

injection/common/lazy.py

@@ -1,52 +1,52 @@
 from collections.abc import Callable, Iterator, Mapping
+from contextlib import suppress
 from types import MappingProxyType
 from typing import Any, Generic, TypeVar
 
 from injection.common.tools.threading import thread_lock
 
 __all__ = ("Lazy", "LazyMapping")
 
-_sentinel = object()
-
 _T = TypeVar("_T")
 _K = TypeVar("_K")
 _V = TypeVar("_V")
 
 
 class Lazy(Generic[_T]):
-    __slots__ = ("__factory", "__value")
+    __slots__ = ("is_set", "__generator")
 
     def __init__(self, factory: Callable[[], _T]):
-        self.__factory = factory
-        self.__value = _sentinel
+        def generator() -> Iterator[_T]:
+            nonlocal factory
 
-    def __invert__(self) -> _T:
-        if not self.is_set:
             with thread_lock:
-                self.__value = self.__factory()
-                self.__factory = _sentinel
+                value = factory()
+                self.is_set = True
+                del factory
 
-        return self.__value
+            while True:
+                yield value
+
+        self.is_set = False
+        self.__generator = generator()
+
+    def __invert__(self) -> _T:
+        return next(self.__generator)
+
+    def __call__(self) -> _T:
+        return ~self
 
     def __setattr__(self, name: str, value: Any, /):
-        if self.is_set:
-            raise TypeError(f"`{self}` is frozen.")
+        with suppress(AttributeError):
+            if self.is_set:
+                raise TypeError(f"`{self}` is frozen.")
 
         return super().__setattr__(name, value)
 
-    @property
-    def is_set(self) -> bool:
-        try:
-            return self.__factory is _sentinel
-        except AttributeError:
-            return False
-
 
 class LazyMapping(Mapping[_K, _V]):
     __slots__ = ("__lazy",)
 
-    __lazy: Lazy[MappingProxyType[_K, _V]]
-
     def __init__(self, iterator: Iterator[tuple[_K, _V]]):
         self.__lazy = Lazy(lambda: MappingProxyType(dict(iterator)))
 
@@ -58,3 +58,7 @@ def __iter__(self) -> Iterator[_K]:
 
     def __len__(self) -> int:
         return len(~self.__lazy)
+
+    @property
+    def is_set(self) -> bool:
+        return self.__lazy.is_set

injection/core/module.py

@@ -16,11 +16,17 @@
 from contextlib import ContextDecorator, contextmanager, suppress
 from dataclasses import dataclass, field
 from enum import Enum, auto
-from functools import partialmethod, singledispatchmethod, wraps
+from functools import (
+    partialmethod,
+    singledispatchmethod,
+    update_wrapper,
+    wraps,
+)
 from inspect import Signature, isclass
-from types import MappingProxyType, UnionType
+from types import UnionType
 from typing import (
     Any,
+    ClassVar,
     ContextManager,
     NamedTuple,
     NoReturn,
@@ -169,7 +175,7 @@ def get_instance(self) -> _T:
 class SingletonInjectable(BaseInjectable[_T]):
     __slots__ = ("__dict__",)
 
-    __INSTANCE_KEY = "$instance"
+    __INSTANCE_KEY: ClassVar[str] = "$instance"
 
     @property
     def cache(self) -> MutableMapping[str, Any]:
@@ -405,21 +411,15 @@ def inject(
         wrapped: Callable[..., Any] = None,
         /,
         *,
-        force: bool = False,
         return_factory: bool = False,
     ):
         def decorator(wp):
             if not return_factory and isclass(wp):
-                wp.__init__ = self.inject(wp.__init__, force=force)
+                wp.__init__ = self.inject(wp.__init__)
                 return wp
 
-            lazy_binder = Lazy[Binder](lambda: self.__new_binder(wp))
-
-            @wraps(wp)
-            def wrapper(*args, **kwargs):
-                arguments = (~lazy_binder).bind(args, kwargs, force)
-                return wp(*arguments.args, **arguments.kwargs)
-
+            wrapper = InjectedFunction(wp).update(self)
+            self.add_listener(wrapper)
             return wrapper
 
         return decorator(wrapped) if wrapped else decorator
@@ -535,21 +535,15 @@ def __move_module(self, module: Module, priority: ModulePriority):
                 f"`{module}` can't be found in the modules used by `{self}`."
             ) from exc
 
-    def __new_binder(self, target: Callable[..., Any]) -> Binder:
-        signature = inspect.signature(target, eval_str=True)
-        binder = Binder(signature).update(self)
-        self.add_listener(binder)
-        return binder
-
 
 """
-Binder
+InjectedFunction
 """
 
 
 @dataclass(repr=False, frozen=True, slots=True)
 class Dependencies:
-    mapping: MappingProxyType[str, Injectable]
+    mapping: Mapping[str, Injectable]
 
     def __bool__(self) -> bool:
         return bool(self.mapping)
@@ -558,75 +552,137 @@ def __iter__(self) -> Iterator[tuple[str, Any]]:
         for name, injectable in self.mapping.items():
             yield name, injectable.get_instance()
 
+    @property
+    def are_resolved(self) -> bool:
+        if isinstance(self.mapping, LazyMapping) and not self.mapping.is_set:
+            return False
+
+        return bool(self)
+
     @property
     def arguments(self) -> OrderedDict[str, Any]:
         return OrderedDict(self)
 
     @classmethod
     def from_mapping(cls, mapping: Mapping[str, Injectable]):
-        return cls(mapping=MappingProxyType(mapping))
+        return cls(mapping=mapping)
 
     @classmethod
     def empty(cls):
         return cls.from_mapping({})
 
     @classmethod
-    def resolve(cls, signature: Signature, module: Module):
-        dependencies = LazyMapping(cls.__resolver(signature, module))
+    def resolve(cls, signature: Signature, module: Module, owner: type = None):
+        dependencies = LazyMapping(cls.__resolver(signature, module, owner))
         return cls.from_mapping(dependencies)
 
     @classmethod
     def __resolver(
         cls,
         signature: Signature,
         module: Module,
+        owner: type = None,
     ) -> Iterator[tuple[str, Injectable]]:
-        for name, parameter in signature.parameters.items():
+        for name, annotation in cls.__get_annotations(signature, owner):
             try:
-                injectable = module[parameter.annotation]
+                injectable = module[annotation]
             except KeyError:
                 continue
 
             yield name, injectable
 
+    @staticmethod
+    def __get_annotations(
+        signature: Signature,
+        owner: type = None,
+    ) -> Iterator[tuple[str, type | Any]]:
+        parameters = iter(signature.parameters.items())
+
+        if owner:
+            name, _ = next(parameters)
+            yield name, owner
+
+        for name, parameter in parameters:
+            yield name, parameter.annotation
+
 
 class Arguments(NamedTuple):
     args: Iterable[Any]
     kwargs: Mapping[str, Any]
 
 
-class Binder(EventListener):
-    __slots__ = ("__signature", "__dependencies")
+class InjectedFunction(EventListener):
+    __slots__ = ("__dict__", "__wrapper", "__dependencies", "__owner")
+
+    def __init__(self, wrapped: Callable[..., Any], /):
+        update_wrapper(self, wrapped)
+        self.__signature__ = Lazy[Signature](
+            lambda: inspect.signature(wrapped, eval_str=True)
+        )
+
+        @wraps(wrapped)
+        def wrapper(*args, **kwargs):
+            args, kwargs = self.bind(args, kwargs)
+            return wrapped(*args, **kwargs)
 
-    def __init__(self, signature: Signature):
-        self.__signature = signature
+        self.__wrapper = wrapper
         self.__dependencies = Dependencies.empty()
+        self.__owner = None
+
+    def __repr__(self) -> str:
+        return repr(self.__wrapper)
+
+    def __str__(self) -> str:
+        return str(self.__wrapper)
+
+    def __call__(self, /, *args, **kwargs) -> Any:
+        return self.__wrapper(*args, **kwargs)
+
+    def __get__(self, instance: object | None, owner: type):
+        if instance is None:
+            return self
+
+        return self.__wrapper.__get__(instance, owner)
+
+    def __set_name__(self, owner: type, name: str):
+        if self.__dependencies.are_resolved:
+            raise TypeError(
+                "`__set_name__` is called after dependencies have been resolved."
+            )
+
+        if self.__owner:
+            raise TypeError("Function owner is already defined.")
+
+        self.__owner = owner
+
+    @property
+    def signature(self) -> Signature:
+        return self.__signature__()
 
     def bind(
         self,
         args: Iterable[Any] = (),
         kwargs: Mapping[str, Any] = None,
-        force: bool = False,
     ) -> Arguments:
         if kwargs is None:
             kwargs = {}
 
         if not self.__dependencies:
             return Arguments(args, kwargs)
 
-        bound = self.__signature.bind_partial(*args, **kwargs)
+        bound = self.signature.bind_partial(*args, **kwargs)
         dependencies = self.__dependencies.arguments
-
-        if force:
-            bound.arguments |= dependencies
-        else:
-            bound.arguments = dependencies | bound.arguments
+        bound.arguments = dependencies | bound.arguments
 
         return Arguments(bound.args, bound.kwargs)
 
     def update(self, module: Module):
         with thread_lock:
-            self.__dependencies = Dependencies.resolve(self.__signature, module)
+            self.__dependencies = Dependencies.resolve(
+                self.signature,
+                module,
+                self.__owner,
+            )
 
         return self