refactor: Extract DI logic into DIAdapter protocol

Author: remimd

Makefile

@@ -1,11 +1,11 @@
 before_commit: lint mypy pytest
 
 install:
-	uv sync
+	uv sync --all-extras
 
 update:
 	uv lock --upgrade
-	uv sync
+	uv sync --all-extras
 
 lint:
 	uv run ruff format

conftest.py

@@ -4,9 +4,16 @@
 import pytest
 from injection.testing import load_test_profile, set_test_constant
 
-from cq import Bus, CommandBus, EventBus, QueryBus
+from cq import (
+    Bus,
+    CommandBus,
+    EventBus,
+    QueryBus,
+    new_command_bus,
+    new_event_bus,
+    new_query_bus,
+)
 from cq._core.dispatcher.bus import SimpleBus
-from cq._core.message import new_command_bus, new_event_bus, new_query_bus
 from tests.helpers.history import HistoryMiddleware
 
 

cq/__init__.py

@@ -1,6 +1,8 @@
-from ._core.dispatcher.base import DeferredDispatcher, Dispatcher
+from ._core.cq import CQ
+from ._core.di import DIAdapter
+from ._core.di import NoDI as _NoDI
+from ._core.dispatcher.base import Dispatcher
 from ._core.dispatcher.bus import Bus
-from ._core.dispatcher.lazy import LazyDispatcher
 from ._core.dispatcher.pipe import ContextPipeline, Pipe
 from ._core.message import (
     AnyCommandBus,
@@ -10,31 +12,24 @@
     EventBus,
     Query,
     QueryBus,
-    command_handler,
-    event_handler,
-    new_command_bus,
-    new_event_bus,
-    new_query_bus,
-    query_handler,
 )
 from ._core.middleware import Middleware, MiddlewareResult, resolve_handler_source
-from ._core.pipetools import ContextCommandPipeline
-from ._core.related_events import RelatedEvents
-from ._core.scope import CQScope
+from ._core.pipetools import ContextCommandPipeline as _ContextCommandPipeline
+from ._core.related_events import AnyIORelatedEvents, RelatedEvents
 
 __all__ = (
     "AnyCommandBus",
+    "AnyIORelatedEvents",
     "Bus",
-    "CQScope",
+    "CQ",
     "Command",
     "CommandBus",
     "ContextCommandPipeline",
     "ContextPipeline",
-    "DeferredDispatcher",
+    "DIAdapter",
     "Dispatcher",
     "Event",
     "EventBus",
-    "LazyDispatcher",
     "Middleware",
     "MiddlewareResult",
     "Pipe",
@@ -49,3 +44,32 @@
     "query_handler",
     "resolve_handler_source",
 )
+
+try:
+    from cq.ext.injection import InjectionAdapter as _InjectionAdapter
+
+except ImportError:  # pragma: no cover
+    _default = CQ(_NoDI())
+
+else:
+    _default = CQ(_InjectionAdapter())
+
+_default.register_defaults()
+
+command_handler = _default.command_handler
+event_handler = _default.event_handler
+query_handler = _default.query_handler
+
+new_command_bus = _default.new_command_bus
+new_event_bus = _default.new_event_bus
+new_query_bus = _default.new_query_bus
+
+
+class ContextCommandPipeline[C: Command](_ContextCommandPipeline[C]):
+    __slots__ = ()
+
+    def __init__(self, di: DIAdapter = _default.di) -> None:
+        super().__init__(di)
+
+
+del _default

cq/_core/cq.py

@@ -0,0 +1,63 @@
+from typing import Any, Self
+
+from cq._core.di import DIAdapter
+from cq._core.dispatcher.bus import Bus, SimpleBus, TaskBus
+from cq._core.handler import (
+    HandlerDecorator,
+    HandlerRegistry,
+    MultipleHandlerRegistry,
+    SingleHandlerRegistry,
+)
+from cq._core.message import Command, Event, Query
+from cq._core.middlewares.scope import CommandDispatchScopeMiddleware
+
+
+class CQ:
+    __slots__ = ("__command_registry", "__di", "__event_registry", "__query_registry")
+
+    __command_registry: HandlerRegistry[Command, Any]
+    __di: DIAdapter
+    __event_registry: HandlerRegistry[Event, Any]
+    __query_registry: HandlerRegistry[Query, Any]
+
+    def __init__(self, di: DIAdapter, /) -> None:
+        self.__di = di
+        self.__command_registry = SingleHandlerRegistry()
+        self.__event_registry = MultipleHandlerRegistry()
+        self.__query_registry = SingleHandlerRegistry()
+
+    @property
+    def di(self) -> DIAdapter:
+        return self.__di
+
+    @property
+    def command_handler(self) -> HandlerDecorator[Command, Any]:
+        return HandlerDecorator(self.__command_registry, self.__di)
+
+    @property
+    def event_handler(self) -> HandlerDecorator[Event, Any]:
+        return HandlerDecorator(self.__event_registry, self.__di)
+
+    @property
+    def query_handler(self) -> HandlerDecorator[Query, Any]:
+        return HandlerDecorator(self.__query_registry, self.__di)
+
+    def new_command_bus(self) -> Bus[Command, Any]:
+        bus = SimpleBus(self.__command_registry)
+        command_middleware = CommandDispatchScopeMiddleware(self.__di)
+        bus.add_middlewares(command_middleware)
+        return bus
+
+    def new_event_bus(self) -> Bus[Event, None]:
+        return TaskBus(self.__event_registry)
+
+    def new_query_bus(self) -> Bus[Query, Any]:
+        return SimpleBus(self.__query_registry)
+
+    def register_defaults(self) -> Self:
+        self.__di.register_defaults(
+            self.new_command_bus,
+            self.new_event_bus,
+            self.new_query_bus,
+        )
+        return self

cq/_core/di.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from collections.abc import Awaitable, Callable
+from contextlib import nullcontext
+from typing import TYPE_CHECKING, Any, AsyncContextManager, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from cq._core.message import CommandBus, EventBus, QueryBus
+
+
+@runtime_checkable
+class DIAdapter(Protocol):
+    __slots__ = ()
+
+    @abstractmethod
+    def command_scope(self) -> AsyncContextManager[None]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def lazy[T](self, tp: type[T]) -> Callable[[], Awaitable[T]]:
+        raise NotImplementedError
+
+    def register_defaults(
+        self,
+        command_bus: Callable[..., CommandBus[Any]],
+        event_bus: Callable[..., EventBus],
+        query_bus: Callable[..., QueryBus[Any]],
+    ) -> None:
+        return
+
+    @abstractmethod
+    def wire[T](self, tp: type[T]) -> Callable[..., Awaitable[T]]:
+        raise NotImplementedError
+
+
+class NoDI(DIAdapter):
+    __slots__ = ()
+
+    def command_scope(self) -> AsyncContextManager[None]:
+        return nullcontext()
+
+    def lazy[T](self, tp: type[T], /) -> Callable[[], Awaitable[T]]:
+        tp_str = getattr(tp, "__name__", str(tp))
+        raise RuntimeError(
+            f"Can't lazily resolve {tp_str}: no DI container configured."
+        )
+
+    def wire[T](self, tp: type[T], /) -> Callable[..., Awaitable[T]]:
+        async def factory() -> T:
+            return tp()
+
+        return factory

cq/_core/dispatcher/base.py

@@ -15,15 +15,6 @@ async def dispatch(self, input_value: I, /) -> O:
         raise NotImplementedError
 
 
-@runtime_checkable
-class DeferredDispatcher[I](Protocol):
-    __slots__ = ()
-
-    @abstractmethod
-    async def defer(self, input_value: I, /) -> None:
-        raise NotImplementedError
-
-
 class BaseDispatcher[I, O](Dispatcher[I, O], ABC):
     __slots__ = ("__middleware_group",)
 

cq/_core/dispatcher/lazy.py

@@ -1,28 +1,24 @@
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Callable
 from types import GenericAlias
 from typing import TypeAliasType
 
-import injection
-
+from cq._core.di import DIAdapter
 from cq._core.dispatcher.base import Dispatcher
 
 
 class LazyDispatcher[I, O](Dispatcher[I, O]):
-    __slots__ = ("__value",)
+    __slots__ = ("__resolve",)
 
-    __value: Awaitable[Dispatcher[I, O]]
+    __resolve: Callable[[], Awaitable[Dispatcher[I, O]]]
 
     def __init__(
         self,
         dispatcher_type: type[Dispatcher[I, O]] | TypeAliasType | GenericAlias,
         /,
-        *,
-        injection_module: injection.Module | None = None,
-        threadsafe: bool | None = None,
+        di: DIAdapter,
     ) -> None:
-        module = injection_module or injection.mod()
-        self.__value = module.aget_lazy_instance(dispatcher_type, threadsafe=threadsafe)
+        self.__resolve = di.lazy(dispatcher_type)  # type: ignore[arg-type]
 
     async def dispatch(self, input_value: I, /) -> O:
-        dispatcher = await self.__value
+        dispatcher = await self.__resolve()
         return await dispatcher.dispatch(input_value)

cq/_core/handler.py

@@ -7,10 +7,10 @@
 from inspect import signature as inspect_signature
 from typing import TYPE_CHECKING, Any, Protocol, Self, overload, runtime_checkable
 
-import injection
 from type_analyzer import MatchingTypesConfig, iter_matching_types, matching_types
 
 from cq._core.common.typing import Decorator
+from cq._core.di import DIAdapter, NoDI
 
 type HandlerType[**P, T] = type[Handler[P, T]]
 type HandlerFactory[**P, T] = Callable[..., Awaitable[Handler[P, T]]]
@@ -126,7 +126,7 @@ def subscribe(
 @dataclass(repr=False, eq=False, frozen=True, slots=True)
 class HandlerDecorator[I, O]:
     registry: HandlerRegistry[I, O]
-    injection_module: injection.Module = field(default_factory=injection.mod)
+    di: DIAdapter = field(default_factory=NoDI)
 
     if TYPE_CHECKING:  # pragma: no cover
 
@@ -137,7 +137,6 @@ def __call__(
             /,
             *,
             fail_silently: bool = ...,
-            threadsafe: bool | None = ...,
         ) -> Decorator: ...
 
         @overload
@@ -147,7 +146,6 @@ def __call__[T](
             /,
             *,
             fail_silently: bool = ...,
-            threadsafe: bool | None = ...,
         ) -> T: ...
 
         @overload
@@ -157,7 +155,6 @@ def __call__(
             /,
             *,
             fail_silently: bool = ...,
-            threadsafe: bool | None = ...,
         ) -> Decorator: ...
 
     def __call__[T](
@@ -166,7 +163,6 @@ def __call__[T](
         /,
         *,
         fail_silently: bool = False,
-        threadsafe: bool | None = None,
     ) -> Any:
         if (
             input_or_handler_type is not None
@@ -176,14 +172,12 @@ def __call__[T](
             return self.__decorator(
                 input_or_handler_type,
                 fail_silently=fail_silently,
-                threadsafe=threadsafe,
             )
 
         return partial(
             self.__decorator,
             input_type=input_or_handler_type,  # type: ignore[arg-type]
             fail_silently=fail_silently,
-            threadsafe=threadsafe,
         )
 
     def __decorator(
@@ -193,9 +187,8 @@ def __decorator(
         *,
         input_type: type[I] | None = None,
         fail_silently: bool = False,
-        threadsafe: bool | None = None,
     ) -> HandlerType[[I], O]:
-        factory = self.injection_module.make_async_factory(wrapped, threadsafe)
+        factory = self.di.wire(wrapped)
         input_type = input_type or _resolve_input_type(wrapped)
         self.registry.subscribe(input_type, factory, wrapped, fail_silently)
         return wrapped