DI Provider ctx #186

src/cpl-dependency/cpl/dependency/__init__.py

@@ -1,3 +1,5 @@
+from .context import get_current_provider, use_provider
+from .inject import inject
 from .scope import Scope
 from .scope_abc import ScopeABC
 from .service_collection import ServiceCollection

src/cpl-dependency/cpl/dependency/context.py

@@ -0,0 +1,21 @@
+import contextvars
+from contextlib import contextmanager
+
+_current_provider = contextvars.ContextVar("current_provider", default=None)
+
+
+def use_root_provider(provider):
+    _current_provider.set(provider)
+
+
+@contextmanager
+def use_provider(provider):
+    token = _current_provider.set(provider)
+    try:
+        yield
+    finally:
+        _current_provider.reset(token)
+
+
+def get_current_provider():
+    return _current_provider.get()

src/cpl-dependency/cpl/dependency/inject.py

@@ -0,0 +1,42 @@
+import functools
+from asyncio import iscoroutinefunction
+from inspect import signature
+
+from cpl.dependency.context import get_current_provider
+
+
+def inject(f=None):
+    if f is None:
+        return functools.partial(inject)
+
+    if iscoroutinefunction(f):
+
+        @functools.wraps(f)
+        async def async_inner(*args, **kwargs):
+            from cpl.dependency.service_provider import ServiceProvider
+
+            provider: ServiceProvider | None = get_current_provider()
+            if provider is None:
+                raise ValueError(
+                    "No provider in current context. Use 'with use_provider(provider):' to set the provider in the current context."
+                )
+
+            injection = [x for x in provider._build_by_signature(signature(f)) if x is not None]
+            return await f(*args, *injection, **kwargs)
+
+        return async_inner
+
+    @functools.wraps(f)
+    def inner(*args, **kwargs):
+        from cpl.dependency.service_provider import ServiceProvider
+
+        provider: ServiceProvider | None = get_current_provider()
+        if provider is None:
+            raise ValueError(
+                "No provider in current context. Use 'with use_provider(provider):' to set the provider in the current context."
+            )
+
+        injection = [x for x in provider._build_by_signature(signature(f)) if x is not None]
+        return f(*args, *injection, **kwargs)
+
+    return inner

src/cpl-dependency/cpl/dependency/service_provider_abc.py

@@ -1,9 +1,8 @@
-import functools
 from abc import abstractmethod, ABC
-from inspect import Signature, signature, iscoroutinefunction
+from inspect import Signature
 from typing import Optional, Type
 
-from cpl.core.typing import T, R
+from cpl.core.typing import T
 from cpl.dependency.scope_abc import ScopeABC
 
 
@@ -126,40 +125,3 @@ class ServiceProviderABC(ABC):
         -------
             Object of type list[:class:`type`]
         """
-
-    @classmethod
-    def inject(cls, f=None):
-        r"""Decorator to allow injection into static and class methods
-
-        Parameter
-        ---------
-            f: Callable
-
-        Returns
-        -------
-            function
-        """
-        if f is None:
-            return functools.partial(cls.inject)
-
-        if iscoroutinefunction(f):
-
-            @functools.wraps(f)
-            async def async_inner(*args, **kwargs):
-                if cls._provider is None:
-                    raise Exception(f"{cls.__name__} not build!")
-
-                injection = [x for x in cls._provider._build_by_signature(signature(f)) if x is not None]
-                return await f(*args, *injection, **kwargs)
-
-            return async_inner
-
-        @functools.wraps(f)
-        def inner(*args, **kwargs):
-            if cls._provider is None:
-                raise Exception(f"{cls.__name__} not build!")
-
-            injection = [x for x in cls._provider._build_by_signature(signature(f)) if x is not None]
-            return f(*args, *injection, **kwargs)
-
-        return inner