Add tags (#79)

* Add tags

* bump version

CHANGELOG.md

@@ -4,7 +4,12 @@ We follow [Semantic Versions](https://semver.org/).
 
 ## Version next
 
--
+## Version 0.23.0
+
+- Now you can mark you dependencies with tags for more granular control over their lifecycle
+- Breaking changes:
+  - Removed `ignore_manual_init` argument from `dependency` decorator
+  - `lifespan` now has new signature
 
 ## Version 0.22.0
 

docs/conf.py

@@ -61,6 +61,8 @@
     "~picodi._scopes.SingletonScope": "SingletonScope",
     "<class 'picodi._scopes.ManualScope'>": "ManualScope",
     "<class 'picodi._scopes.NullScope'>": "NullScope",
+    "<class 'picodi._scopes.SingletonScope'>": "SingletonScope",
+    "<class 'picodi._scopes.ContextVarScope'>": "ContextVarScope",
 }
 
 

docs/scopes.rst

@@ -127,31 +127,30 @@ scope you can inject it in sync code. Example:
 Because ``get_async_dependency`` is ``SingletonScope`` scoped dependency and
 it's initialized on startup, while your app is running you can inject it in sync code.
 
-Skipping manual initialization
-******************************
+Initializing dependencies by tags
+*********************************
 
-If you want to skip manual initialization of your dependency you can use the
-``ignore_manual_init`` argument of the :func:`picodi.dependency` decorator.
+You can tag your dependencies to initialize them selectively.
 
 .. testcode::
 
     from picodi import SingletonScope, dependency, init_dependencies
 
 
-    # Try to set `ignore_manual_init` to `False` and see what happens
-    @dependency(scope_class=SingletonScope, ignore_manual_init=True)
-    def get_dependency():
-        print("This will not be printed")
-        return "from async"
+    @dependency(scope_class=SingletonScope, tags=["tag1"])
+    def get_tagged_dependency():
+        return "tagged dependency"
 
 
-    init_dependencies()  # This will not initialize get_dependency
+    init_dependencies(tags=["tag1"])
+    # or you can exclude dependencies by tags
+    init_dependencies(tags=["-tag1"])
 
 Managing scopes selectively
 ***************************
 
-By default :func:`picodi.init_dependencies` and :func:`picodi.shutdown_dependencies`
-will initialize and close all dependencies with :class:`SingletonScope`.
+By default :func:`picodi.init_dependencies` initialize
+dependencies with :class:`SingletonScope`.
 If you want to manage scopes selectively you can use the
 ``scope_class`` argument of these functions.
 

picodi/_picodi.py

@@ -47,6 +47,8 @@
 
 
 DependencyCallable = Callable[..., Any]
+LifespanScopeClass = type[ManualScope] | tuple[type[ManualScope], ...]
+Tags = list[str] | tuple[str, ...]
 T = TypeVar("T")
 P = ParamSpec("P")
 TC = TypeVar("TC", bound=Callable)
@@ -73,7 +75,7 @@ def add(
         dependency: DependencyCallable,
         scope_class: type[ScopeType] = NullScope,
         override_scope: bool = False,
-        ignore_manual_init: bool | Callable[[], bool] = False,
+        tags: Tags = (),
     ) -> None:
         """
         Add a dependency to the registry.
@@ -90,7 +92,7 @@ def add(
                 self._storage.deps[dependency] = Provider.from_dependency(
                     dependency=dependency,
                     scope_class=scope_class,
-                    ignore_manual_init=ignore_manual_init,
+                    tags=tags,
                 )
 
     def get(self, dependency: DependencyCallable) -> Provider:
@@ -354,7 +356,7 @@ def gen_wrapper(
 def dependency(
     *,
     scope_class: type[ScopeType] = NullScope,
-    ignore_manual_init: bool | Callable[[], bool] = False,
+    tags: Tags = (),
 ) -> Callable[[TC], TC]:
     """
     Decorator to declare a dependency. You don't need to use it with default arguments,
@@ -365,11 +367,9 @@ def dependency(
         :class:`NullScope`.
         Picodi additionally provides a few built-in scopes:
         :class:`SingletonScope`, :class:`ContextVarScope`.
-    :param ignore_manual_init: this parameter can be used to skip the dependency manual
-        initialization. It can be a boolean or a callable that returns a boolean.
-        If it's a callable, it will be called every time before the dependency is
-        initialized with :func:`init_dependencies`. If it returns True, the dependency
-        will be skipped.
+    :param tags: list of tags to assign to the dependency. Tags can be used to select
+        which dependencies to initialize. They can be used to group
+        dependencies and control their lifecycle.
     """
 
     if scope_class not in _scopes:
@@ -380,15 +380,16 @@ def decorator(fn: TC) -> TC:
             fn,
             scope_class=scope_class,
             override_scope=True,
-            ignore_manual_init=ignore_manual_init,
+            tags=tags,
         )
         return fn
 
     return decorator
 
 
 def init_dependencies(
-    scope_class: type[ManualScope] | tuple[type[ManualScope], ...] = SingletonScope,
+    scope_class: LifespanScopeClass = SingletonScope,
+    tags: Tags = (),
 ) -> Awaitable:
     """
     Call this function to close dependencies. Usually, it should be called
@@ -399,14 +400,21 @@ def init_dependencies(
     If you call it ``await init_dependencies()``, it will initialize both sync and async
     dependencies.
 
+    If you not pass any arguments, it will initialize only :class:`SingletonScope`
+    and its subclasses.
+
+    If you pass tags with scope_class ``and`` logic will be applied.
+
     :param scope_class: you can specify the scope class to initialize. If passed -
         only dependencies of this scope class and its subclasses will be initialized.
-        By default, it will initialize only :class:`SingletonScope` and its subclasses.
+    :param tags: you can specify the dependencies to initialize by tags. If passed -
+        only dependencies of this tags will be initialized. If you pass a tag with a
+        minus sign, it will exclude dependencies with this tag.
     """
-    async_deps = []
     filtered_providers = _internal_registry.filter(
-        lambda p: not p.is_ignored() and issubclass(p.scope_class, scope_class)
+        lambda p: p.match_tags(tags) and issubclass(p.scope_class, scope_class)
     )
+    async_deps = []
     for provider in filtered_providers:
         resolver = LazyResolver(provider)
         value = resolver(provider.is_async)
@@ -428,7 +436,7 @@ async def init_all() -> None:
 
 
 def shutdown_dependencies(
-    scope_class: type[ManualScope] | tuple[type[ManualScope], ...] = SingletonScope,
+    scope_class: LifespanScopeClass = ManualScope,
 ) -> Awaitable:
     """
     Call this function to close dependencies. Usually, it should be called
@@ -439,9 +447,10 @@ def shutdown_dependencies(
     If you call it ``await shutdown_dependencies()``, it will shutdown both
     sync and async dependencies.
 
+    If you not pass any arguments, it will shutdown subclasses of :class:`ManualScope`.
+
     :param scope_class: you can specify the scope class to shutdown. If passed -
         only dependencies of this scope class and its subclasses will be shutdown.
-        By default, it will shutdown only :class:`SingletonScope` and its subclasses.
     """
     tasks = [
         instance.shutdown()  # type: ignore[call-arg]
@@ -462,29 +471,34 @@ class Provider:
     dependency: DependencyCallable
     is_async: bool
     scope_class: type[ScopeType]
-    ignore_manual_init: bool | Callable[[], bool]
+    tags: set[str]
 
     @classmethod
     def from_dependency(
         cls,
         dependency: DependencyCallable,
         scope_class: type[ScopeType],
-        ignore_manual_init: bool | Callable[[], bool] = False,
+        tags: Tags,
     ) -> Provider:
         is_async = inspect.iscoroutinefunction(
             dependency
         ) or inspect.isasyncgenfunction(dependency)
+
         return cls(
             dependency=dependency,
             is_async=is_async,
             scope_class=scope_class,
-            ignore_manual_init=ignore_manual_init,
+            tags=set(tags),
         )
 
-    def is_ignored(self) -> bool:
-        if callable(self.ignore_manual_init):
-            return self.ignore_manual_init()
-        return self.ignore_manual_init
+    def match_tags(self, tags: Tags) -> bool:
+        include_tags = {tag for tag in tags if not tag.startswith("-")}
+        exclude_tags = {tag[1:] for tag in tags if tag.startswith("-")}
+
+        if exclude_tags.intersection(self.tags):
+            return False
+
+        return bool(not include_tags or include_tags.intersection(self.tags))
 
     def replace(self, scope_class: type[ScopeType] | None = None) -> Provider:
         kwargs = asdict(self)

picodi/helpers.py

@@ -19,7 +19,7 @@
 
 import picodi
 from picodi import ManualScope, SingletonScope
-from picodi._picodi import _internal_registry
+from picodi._picodi import LifespanScopeClass, Tags, _internal_registry
 from picodi.support import nullcontext
 
 if TYPE_CHECKING:
@@ -146,81 +146,101 @@ def __call__(
         self,
         fn: None = None,
         *,
-        scope_class: type[ManualScope] | tuple[type[ManualScope]] = SingletonScope,
-        skip_init: bool = False,
+        init_scope_class: LifespanScopeClass | None = SingletonScope,
+        init_tags: Tags = (),
+        shutdown_scope_class: LifespanScopeClass | None = ManualScope,
     ) -> Callable[[Callable[P, T]], Callable[P, T]]:
         """Sync and Async context manager"""
 
     def __call__(
         self,
         fn: Callable[P, T] | None = None,
         *,
-        scope_class: type[ManualScope] | tuple[type[ManualScope]] = SingletonScope,
-        skip_init: bool = False,
+        init_scope_class: LifespanScopeClass | None = SingletonScope,
+        init_tags: Tags = (),
+        shutdown_scope_class: LifespanScopeClass | None = ManualScope,
     ) -> Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]:
         """
         Can be used as a decorator or a context manager.
 
         :param fn: function to decorate (if used as a decorator).
-        :param scope_class: optionally you can specify the scope class
-            to initialize and shutdown.
-        :param skip_init: if True, don't initialize dependencies.
+        :param init_scope_class: scope class for initialization
+            (can be omitted by passing None).
+        :param init_tags: list of tags to be initialized.
+        :param shutdown_scope_class: scope class for shutdown
+            (can be omitted by passing None).
         :return: decorated function or decorator.
         """
 
         def decorator(fn: Callable[P, T]) -> Callable[P, T]:
             if asyncio.iscoroutinefunction(fn):
                 return self.async_(  # type: ignore[return-value]
-                    scope_class=scope_class, skip_init=skip_init
+                    init_scope_class=init_scope_class,
+                    init_tags=init_tags,
+                    shutdown_scope_class=shutdown_scope_class,
                 )(fn)
-            return self.sync(scope_class=scope_class, skip_init=skip_init)(fn)
+            return self.sync(
+                init_scope_class=init_scope_class,
+                init_tags=init_tags,
+                shutdown_scope_class=shutdown_scope_class,
+            )(fn)
 
         return decorator if fn is None else decorator(fn)
 
     @contextlib.contextmanager
     def sync(
         self,
         *,
-        scope_class: type[ManualScope] | tuple[type[ManualScope]] = SingletonScope,
-        skip_init: bool = False,
+        init_scope_class: LifespanScopeClass | None = SingletonScope,
+        init_tags: Tags = (),
+        shutdown_scope_class: LifespanScopeClass | None = ManualScope,
     ) -> Generator[None, None, None]:
         """
         :attr:`lifespan` can automatically detect if the decorated function
         is async or not. But if you want to force sync behavior, ``lifespan.sync``.
 
-        :param scope_class: optionally you can specify the scope class
-            to initialize and shutdown.
-        :param skip_init: if True, don't initialize dependencies.
+        :param init_scope_class: scope class for initialization
+            (can be omitted by passing None).
+        :param init_tags: list of tags to be initialized.
+        :param shutdown_scope_class: scope class for shutdown
+            (can be omitted by passing None).
         """
-        if not skip_init:
-            picodi.init_dependencies(scope_class)
+        if init_scope_class is not None:
+            picodi.init_dependencies(init_scope_class, tags=init_tags)
         try:
             yield
         finally:
-            picodi.shutdown_dependencies(scope_class)
+            if shutdown_scope_class is not None:
+                picodi.shutdown_dependencies(shutdown_scope_class)
 
     @contextlib.asynccontextmanager
     async def async_(
         self,
         *,
-        scope_class: type[ManualScope] | tuple[type[ManualScope]] = SingletonScope,
-        skip_init: bool = False,
+        init_scope_class: LifespanScopeClass | None = SingletonScope,
+        init_tags: Tags = (),
+        shutdown_scope_class: LifespanScopeClass | None = ManualScope,
     ) -> AsyncGenerator[None, None]:
         """
         :attr:`lifespan` can automatically detect if the decorated function
         is async or not.
         But if you want to force async behavior, ``lifespan.async_``.
 
-        :param scope_class: optionally you can specify the scope class
-            to initialize and shutdown.
-        :param skip_init: if True, don't initialize dependencies.
+        :param init_scope_class: scope class for initialization
+            (can be omitted by passing None).
+        :param init_tags: list of tags to be initialized.
+        :param shutdown_scope_class: scope class for shutdown
+            (can be omitted by passing None).
         """
-        if not skip_init:
-            await picodi.init_dependencies(scope_class)
+        if init_scope_class is not None:
+            await picodi.init_dependencies(init_scope_class, tags=init_tags)
         try:
             yield
         finally:
-            await picodi.shutdown_dependencies(scope_class)  # noqa: ASYNC102
+            if shutdown_scope_class is not None:
+                await picodi.shutdown_dependencies(  # noqa: ASYNC102
+                    shutdown_scope_class
+                )
 
 
 lifespan = _Lifespan()

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "picodi"
 description = "Simple Dependency Injection for Python"
-version = "0.22.0"
+version = "0.23.0"
 license = "MIT"
 authors = [
   "yakimka"