76class Container:
 77    """
 78    container is a box that can store given factories of types but cannot create
 79    instances, container can be thought as a type factory for a `Scope`, if a scope need to create
 80    a type, it fetches the registered factory for the required type and uses
 81    that factory to create the type
 82
 83    ### THREAD SAFETY
 84        containers are thread safe, they can be used in multiple different places that use
 85        threading, for example, multiple scopes that run on different threads
 86    """
 87
 88    def __init__(self) -> None:
 89        self._factories: PrefixTree[type, Factory] = PrefixTree(
 90            find_strategy=PrefixTreeTypeFindStrategy()
 91        )
 92        self._partially_initialized: list[Factory] = []
 93        self._lock = threading.Lock()
 94        self._setup()
 95
 96    def has_registered(self, type_: Any) -> bool:
 97        """
 98        returns a boolean value indicating if the type was registered
 99        for the current container
100        """
101        return self._get_factory(type_) is not None
102
103    def _register(self, factory: Factory) -> None:
104        """
105        registers the given factory with the scope, the factory return type
106        will be used as key when calling the container `get_factory` and providing a type
107        """
108        is_partial = _is_partial_factory(factory)
109        with self._lock:
110            if is_partial:
111                self._factories.insert(
112                    type_as_prefix_steps(factory.return_type), factory
113                )
114            else:
115                self._partially_initialized.append(factory)
116
117    def update_forward_refs(self, module: ModuleType) -> None:
118        # TODO: should we call `insepct.currentframe` to have visibility
119        # for locals?
120        with self._lock:
121            self._update_forward_ref(module)
122
123    def _update_forward_ref(self, module: ModuleType) -> None:
124        partially_initialized = []
125
126        for factory in self._partially_initialized:
127            if factory.module is module:
128                self._factories.insert(
129                    type_as_prefix_steps(factory.return_type),
130                    _evaluate_partial_factory(factory),
131                )
132            else:
133                partially_initialized.append(factory)
134        self._partially_initialized = partially_initialized
135
136    def _get_factory(
137        self, type_: type | GenericAlias | TypeAliasType
138    ) -> Factory | None:
139        """
140        returns the correct factory for the requested type
141        """
142        prefix = type_as_prefix_steps(type_)
143        with self._lock:
144            return self._factories.find(prefix)
145
146    def _setup(self) -> None:
147        from ._builtins import Lazy
148
149        Injectable(self).register(Lazy)

 38class Scope:
 39    """
 40    hold the live instances, different scope do not share instances
 41    between them unless explicitly annotated via `cdi.InjectableMetadata(provider_scope=...)`
 42
 43    live instances can be inserted into the scope more about it look at `Scope.insert_instance`,
 44    the scope cannot manipulate the bounded container in any way
 45
 46    scopes can inherit from different scopes, aquiring those roles do not effect the scope
 47    behavior, it is more for conviniance when using `cdi.InjectableMetadata(provider_scope=...)`
 48
 49    the scope name is mostly for easier debugging and clearer errors
 50
 51    ## evaluation
 52    the scope uses the given `Container` to look up for factories that can produce the desired type
 53    if the container has no such type the type evaluation will be moved to the provided `no_factory_policy`
 54    that is called when a type has no factory and continue the evaluation from there
 55
 56    by default the policy is `cdi.policy.ErrorNoFactoryPolicy()`
 57
 58    ## thread safety
 59    scope is threadsafe, for every instance an internal lock is aquired preventing
 60    2 threads evaluating at the same time
 61
 62    if the evaluation requires the parent, and no evaluation operation is required on the child scope
 63    then the child scope lock will not be acquired
 64    """
 65
 66    def __init__(
 67        self,
 68        __name: str,
 69        /,
 70        *,
 71        container: Container,
 72        parent: Scope | None = None,
 73        no_factory_policy: NoFactoryPolicy | None = None,
 74    ) -> None:
 75        self._name = __name
 76        self._parent = parent
 77        self._container = container
 78        self._instances: PrefixTree[type, Any] = PrefixTree(
 79            find_strategy=PrefixTreeTypeFindStrategy()
 80        )
 81        self._no_factory_policy = no_factory_policy or ErrorNoFactoryPolicy()
 82
 83        self._stack: list[type] = []
 84        self._lock = threading.RLock()
 85        self._setup()
 86
 87    @property
 88    def name(self) -> str:
 89        return self._name
 90
 91    @property
 92    def container(self) -> Container:
 93        return self._container
 94
 95    @property
 96    def parent(self) -> Scope | None:
 97        """returns the parent of the scope if any"""
 98        return self._parent
 99
100    def fork(self, __name: str | None = None, /) -> Scope:
101        """
102        forks the scope, creating a sub scope that has the same container
103        as the current scope, and set the current scope as the parent
104        """
105        return Scope(
106            __name or (self.name + "-fork"),
107            container=self._container,
108            parent=self,
109            no_factory_policy=self._no_factory_policy,
110        )
111
112    def has_instance(self, __type: Any, /) -> bool:
113        """
114        returns a boolean value indicating if the scope has a live
115        instance of the given type
116
117        it doesn't check if the scope can create such type, for possibility
118        of type creation refer to the scope `cdi.Container` from `scope.container`
119        """
120        prefix = type_as_prefix_steps(__type)
121        with self._lock:
122            return self._instances.find(prefix) is not None
123
124    def get_instance(self, __type: TypeForm[_T], /) -> _T:
125        """
126        returns an instance of the given type, assuming some factory
127        is registered at the container level that can create the type
128        """
129        return self._get_instance_impl(__type, typevars={})
130
131    @contextlib.contextmanager
132    def lifetime(self, label: str | None = None, /) -> Iterator[Scope]:
133        """
134        a context manager that returns a scope, for existing instance, the scope will attempt
135        to get the existing instance
136
137        newly created instances will be bounded to the lifetime scope, so when the lifetime scope
138        is dropped, all the instances created in the lifetime will drop as well
139
140        the lifetime scope will not insert newly created instances into the parent scope
141
142        ```py
143        scope = cdi.Scope(ctr)
144        assert not scope.has_instance(str)
145
146        with scope.lifetime() as lifetime:
147            lifetime.get_instance(str)
148
149        # although `lifetime` called `get_instance`, the `scope` has no
150        # such instance because it was created under the lifetime context
151        assert not scope.has_instance(str)
152        ```
153        """
154        yield Scope(
155            self.name + "-lifetime" + (("-" + label) if label else ""),
156            parent=self,
157            # we create a new container to trick the scope
158            # to always call the `no_factory_policy`
159            container=Container(),
160            no_factory_policy=_LifetimePolicy(),
161        )
162
163    def insert_instance(self, __instance: Any, /) -> None:
164        """
165        inserts the live instance into the scope, mapping between the instance
166        type to the instance itself
167
168        ```py
169        my_obj = object()
170
171        scope = cdi.Scope(...)
172        scope.insert_instance(100)
173        scope.insert_instance(my_obj)
174
175        assert scope.get_instance(int) is 100
176        assert scope.get_instance(object) is my_obj
177        ```
178        """
179        prefix = type_as_prefix_steps(type(__instance))
180        with self._lock:
181            self._instances.insert(prefix, __instance)
182
183    def _get_instance_impl(
184        self,
185        type_: Any,
186        typevars: dict[TypeVar, Any],
187    ) -> Any:
188        """
189        responsible to take any possible type and unwrap it to the real
190        value to check annotations metadata before evaluation
191
192        type evaluation should not be here, but prepare the type
193        for the next step
194        """
195        exceptions = []
196
197        # deque is used to prevent extra stack calls if the `type_` is a
198        # union or annotated, the stack calls can add up for deep injections
199        # with a lot of unions
200        types_: deque[tuple[Any, InjectableMetadata]] = deque()
201        types_.append((type_, InjectableMetadata()))
202
203        while types_:
204            type_, metadata = types_.popleft()
205            origin = get_origin(type_)
206
207            if origin is not None:
208                if typevars:
209                    # if we are in a typealias and we have typevars forwarded
210                    # we should try to resolve possible generics in the current
211                    # `type_` if any, to their real value
212                    type_ = _resolve_type_generics(type_, typevars)
213                if is_union(type_):
214                    types_.extendleft((t, metadata) for t in _unwrap_union(type_))
215                    continue
216                if origin is Annotated:
217                    # look for the relevant metadata and also update the real
218                    # value of the annontated type
219                    inner, annotated_metadata = _get_annotated_injectable_metadata(
220                        type_
221                    )
222                    types_.appendleft(
223                        (
224                            inner,
225                            # try to merge previous metadata with the annotated metadata
226                            # but give more priority to the annotation truthy values
227                            annotated_metadata.merge(metadata)
228                            if annotated_metadata
229                            else metadata,
230                        )
231                    )
232                    continue
233
234            scope = self
235            custom_error_message = None
236
237            try:
238                if metadata._error_message:
239                    custom_error_message = metadata._error_message
240                if metadata._provider_scope is not None:
241                    scope = metadata._provider_scope(self)
242
243                    if scope is not self:
244                        # if need to call a different scope to evaluate the
245                        # type, we don't want the provider scope to call `_provider_scope`
246                        # the results can be unexpected
247                        metadata_copy = copy.copy(metadata)
248                        metadata_copy._provider_scope = None
249
250                        # if the type was wrapped with `Annotated` we want to
251                        # call `_get_instance` to instantiate the real type
252                        return scope._get_instance_impl(
253                            Annotated[type_, metadata_copy],  # type: ignore
254                            typevars=typevars,
255                        )
256
257                dry_instance = self._get_dry_type(type_)
258                if dry_instance is not _miss:
259                    return dry_instance
260                elif metadata._transient:
261                    return scope._create_instance(type_)
262                else:
263                    return scope._get_unwrapped_type(type_)
264            except NoFactoryForTypeError as e:
265                exception = TypeEvaluationError(
266                    f"{self} failed to evaluate type `{type_.__name__}` due to error: "
267                    + str(e)
268                    + "\n"
269                    + (custom_error_message or "")
270                )
271
272                if not types_:
273                    raise exception
274                exceptions.append(exception)
275
276        error_message = (
277            f"{self} failed to evaluate type {type_} due to errors:\n"
278            + "\n - ".join(map(str, exceptions))
279        )
280        raise TypeEvaluationError(error_message)
281
282    def _get_unwrapped_type(self, type_: Any) -> Any:
283        with self._lock:
284            tree_prefix = type_as_prefix_steps(type_)
285            if instance := self._instances.find(tree_prefix):
286                return instance
287
288            if type_ in self._stack:
289                raise CircularDependencyError(tuple(self._stack), type_)
290
291            # TODO: I am pretty sure there is a bug with generic aliases
292            # types since each generic alias is a different instance so in `in` operation will be falsy
293            # check in future
294            self._stack.append(type_)
295
296            try:
297                instance = self._create_instance(type_)
298                self._instances.insert(tree_prefix, instance)
299                return instance
300            finally:
301                popped = self._stack.pop()
302                if popped is not type_:
303                    raise IncorrectStackPopping(
304                        f"{self} incorrect scope stack popping for {self}, expected `{type_.__name__}`, popped `{popped.__name__}`"
305                    )
306
307    def _get_dry_type(self, type_: Any) -> Any:
308        """
309        the term "dry" here means that the type doesn't require touching
310        the `self` (scope) instances, so no locking is required
311        or getting anything from the scope
312
313        this method can be `staticmethod` for all I care, but it is better
314        keeping it here
315        """
316        if type_ is NoneType:
317            return None
318        if type_ is Scope:
319            return self
320
321        if is_generic_alias(type_):
322            type_ = cast(GenericAlias, type_)
323            origin = get_origin(type_)
324
325            if origin is type:
326                return get_args(type_)[0]
327        return _miss
328
329    def _create_instance(self, type_: Any) -> Any:
330        if factory := self._container._get_factory(type_):
331            return self._instantiate_from_factory(factory, _get_typevar_mapping(type_))
332
333        try:
334            return self._no_factory_policy.handle(self, type_)
335        except Exception as e:
336            raise NoFactoryForTypeError(tuple(self._stack), type_) from e
337
338    def _instantiate_from_factory(
339        self, factory: Factory, typevars: dict[TypeVar, Any]
340    ) -> Any:
341        positional_arguments = []
342        keyword_arguments = {}
343
344        for name, parameter in factory.parameters.items():
345            annotation = typevars.get(parameter.annotation, parameter.annotation)
346            value = self._get_instance_impl(annotation, typevars=typevars)
347
348            match parameter.kind:
349                case ParameterKind.POSITIONAL:
350                    positional_arguments.append(value)
351                case ParameterKind.KEYWORD:
352                    keyword_arguments[name] = value
353        return factory(*positional_arguments, **keyword_arguments)
354
355    def _setup(self) -> None:
356        from ._builtins import ContextVar
357
358        self.insert_instance(ContextVar(self.name + "-contextvar"))
359
360    def __str__(self) -> str:
361        return f"Scope<{self.name}>"

 24class Injectable:
 25    """
 26    responsible for creating factories from given types to inject
 27    into the bounded container
 28
 29    the class can be used as standalone or as a decorator
 30    ```py
 31    ctr = cdi.Container()
 32
 33    @cdi.Injectable(ctr)
 34    class Foo:
 35        def __init__(self, name: str) -> None: ...
 36    ```
 37
 38    to read more about how the Injectable behaves for different data types
 39    look at `cdi.Injectable.register`
 40    """
 41
 42    def __init__(self, __ctr: Container, /) -> None:
 43        self._ctr = __ctr
 44
 45    @property
 46    def container(self) -> Container:
 47        """returns the container the injector is bounded to"""
 48        return self._ctr
 49
 50    def register(self, injectable: T) -> None:
 51        """
 52        registers the given type into the provided container, an internal factory
 53        will be generated based on the given type
 54
 55        ### Constant values
 56        when registering an instance (value) and not a class or a function, a wrapper factory
 57        will be built for that instance, calling the factory will always yield the same instance
 58
 59        in a way it is like creating a global instances for a container that any scope
 60        will have access to
 61
 62        ```py
 63        ctr = cdi.Container()
 64
 65        class Foo:
 66            pass
 67
 68        instance = Foo()
 69        cdi.Injectable(ctr).register(instance)
 70
 71        scope1 = cdi.Scope(__name__ + "1", container=ctr)
 72        scope2 = cdi.Scope(__name__ + "2", container=ctr)
 73
 74        # both share the same `instance` of `Foo`
 75        assert scope1.get_instance(Foo) is scope2.get_instance(Foo)
 76        ```
 77
 78        ### Classes
 79        when injecting a class, the parameters will be fetched from the class `__init__`
 80        and on instanciation time, the correct type will be injected based on the type hint,
 81        the factory implementation will call the class `__call__` method
 82
 83        inheritance is supported, including `*args, **kwargs` in your `__init__` class will cause
 84        the injector to move up the parent classes with respect to the `MRO` and evaluate the parent
 85        parameters too
 86
 87        ```py
 88        ctr = cdi.Container()
 89
 90        class Parent:
 91            def __init__(self, parent_field: int) -> None: ...
 92
 93        class Child(Parent):
 94            def __init__(self, child_field: int, *args, **kwargs) -> None: ...
 95
 96        # register int so it will be injected to `parent_field` and `child_field`
 97        cdi.Injectable(ctr).register(100)
 98
 99        # the instance is able to be created without error, `parent_field` and `child_field`
100        # are required fields, and they will be injected
101        cdi.Scope(__name__, container=ctr).get_instance(Child)
102        ```
103
104        FUNCTIONS:
105            parameters and return type are calculated based on the function signature
106        """
107        if inspect.isfunction(injectable):
108            self._inject_func_factory(injectable)
109        elif inspect.isclass(injectable):
110            self._inject_class_factory(injectable)
111        else:
112            self._inject_constant(injectable)
113
114    def __call__(self, injectable: T) -> T:
115        self.register(injectable)
116        return injectable
117
118    def _inject_class_factory(self, cls: type) -> None:
119        parameters = MroParameters().get_parameters(inspect.getmro(cls), "__init__")
120        self._ctr._register(
121            FactoryBuilder()
122            .with_name(f"{cls.__name__}.__init__")
123            .with_func_impl(getattr(cls, "__call__"))
124            .with_module(cast(ModuleType, inspect.getmodule(cls)))
125            .with_parameters(parameters)
126            .with_return_type(cls)
127            .build()
128        )
129
130    def _inject_func_factory(self, func: Callable[..., Any]) -> None:
131        parameters = FuncParameters().get_parameters(func)
132        module = cast(ModuleType, inspect.getmodule(func))
133        rt = inspect.signature(func).return_annotation
134
135        if is_forward_ref(rt):
136            try:
137                rt = ForwardRefResolver(
138                    strategy=ForwardRefResolveByModuleStrategy(module)
139                ).resolve(rt)
140            except ResolveForwardRefError:
141                pass
142
143        self._ctr._register(
144            FactoryBuilder()
145            .with_name(func.__name__)  # type: ignore
146            .with_module(module)
147            .with_parameters(parameters)
148            .with_func_impl(func)
149            .with_return_type(rt)
150            .build()
151        )
152
153    def _inject_constant(self, constant: Any) -> None:
154        self._ctr._register(
155            FactoryBuilder()
156            .with_name(str(constant))
157            .with_func_impl(lambda: constant)
158            .with_return_type(type(constant))
159            .with_module(cast(ModuleType, inspect.getmodule(constant)))
160            .build()
161        )

 99class InjectableMetadata:
100    def __init__(
101        self,
102        *,
103        transient: bool = False,
104        provider_scope: Callable[[Scope], Scope] | None = None,
105        error_message: str | None = None,
106    ) -> None:
107        self._transient = transient
108        self._provider_scope = provider_scope
109        self._error_message = error_message
110
111    def merge(self, other: InjectableMetadata) -> InjectableMetadata:
112        """
113        merges data from `other` into `self`, wherever `self` has a falsy value, the value will
114        be taken from the given `other`
115        """
116        return InjectableMetadata(
117            transient=self._transient or other._transient,
118            provider_scope=self._provider_scope or other._provider_scope,
119            error_message=self._error_message or other._error_message,
120        )
121
122    def __repr__(self) -> str:
123        return (
124            "InjectableMetadata("
125            f"transient={self._transient}, "
126            f"provider_scope={self._provider_scope}, "
127            f"error_message={self._error_message}, "
128            ")"
129        )

 9class Lazy(Generic[_T]):
10    """
11    the lazy type is a builtin generic type that can be used in
12    cases of circular dependency
13
14    ```py
15    @cdi.Injectable(ctr)
16    class A:
17        def __init__(self, b: B) -> None: ...
18
19    @cdi.Injectable(ctr)
20    class B:
21        def __init__(self, a: A) -> None: ...
22
23    ctr.update_forward_refs(sys.modules[__name__])
24
25    scope = cdi.Scope(ctr)
26    scope.get_instance(A)  # error due to circular dep
27    ```
28
29    lazy solves the issue by making one of the dependencies a lazy
30    object
31    ```py
32    @cdi.Injectable(ctr)
33    class A:
34        def __init__(self, b: B) -> None: ...
35
36    @cdi.Injectable(ctr)
37    class B:
38        def __init__(self, a: cdi.Lazy[A]) -> None: ...
39
40    ctr.update_forward_refs(sys.modules[__name__])
41
42    scope = cdi.Scope(ctr)
43    instance = scope.get_instance(A)  # ok
44    assert instance.b.a.wake() is instance
45    ```
46    """
47
48    def __init__(
49        self,
50        type_: type[_T],
51        scope: Annotated[
52            cdi.Scope,
53            cdi.InjectableMetadata(
54                error_message="to use `Lazy` it is required to insert the `Scope` instance"
55            ),
56        ],
57    ) -> None:
58        """@private"""
59        self._type = type_
60        self._scope = scope
61        self._instance: _T | None = None
62
63    def wake(self) -> _T:
64        """wakes up the lazy type for instance evaluation"""
65        if self._instance is None:
66            self._instance = self._scope.get_instance(self._type)
67        return self._instance

12class ContextVar:
13    """
14    injectable representation of python builtin `contextvars`, providing thread local arguments
15    or task local arguments (in async)
16
17    ### memory consumption
18    contextvars has a limition that a `ContextVar` is holding a hard reference to its inner value
19    and when a thread is dropped, the contextvar is not dropped from the global `Context` causing the inner
20    value to never be dropped
21
22    and if the inner value is a list or a dict that holds references, the dict/list items will not be dropped, because
23    of that all values can be set only via `limited` contextmanager, so it will ensure a cleanup inside
24    the contextvar inner dict
25    """
26
27    def __init__(
28        self,
29        name: str,
30    ) -> None:
31        """@private"""
32        self._name = name
33        self._ctx: contextvars.ContextVar[dict[Hashable, Any]] = contextvars.ContextVar(
34            name
35        )
36
37    @property
38    def name(self) -> str:
39        return self._name
40
41    def get(self, key: Hashable, default: _T | None = None) -> Any | _T | None:
42        """
43        tries to return the value for the given key in the thread local contextvar
44        if not found returns `default`
45        """
46        if ctx := self._ctx.get(None):
47            return ctx.get(key, default)
48        return default
49
50    def _insert(self, key: Hashable, value: Any) -> None:
51        current = self._get_value()
52        current[key] = value
53
54    def _delete(self, key: Hashable) -> None:
55        current = self._get_value()
56        current.pop(key, None)
57
58    def _get_value(self) -> dict[Hashable, Any]:
59        if (value := self._ctx.get(None)) is None:
60            value = {}
61            self._ctx.set(value)
62        return value
63
64    @contextlib.contextmanager
65    def limited(self, **kwargs) -> Iterator[None]:
66        """
67        binds the given `kwargs` to the thread local contextvar, when leaving the contextmanager
68        the `kwargs` are cleaned
69        """
70
71        try:
72            for key, value in kwargs.items():
73                self._insert(key, value)
74            yield
75        finally:
76            for key in kwargs:
77                self._delete(key)