From c66a525647e7adf41c085d042765c6b8ce507be2 Mon Sep 17 00:00:00 2001 From: remimd Date: Sun, 30 Aug 2026 14:48:50 +0200 Subject: [PATCH] feat: Add Router.command_pipeline --- cq/__init__.py | 11 ++--------- cq/_core/routing/command_pipeline.py | 11 +++++++++-- cq/_core/routing/router.py | 4 ++++ tests/test_context_command_pipeline.py | 4 +--- 4 files changed, 16 insertions(+), 14 deletions(-) diff --git a/cq/__init__.py b/cq/__init__.py index d34ba8e..e903c08 100644 --- a/cq/__init__.py +++ b/cq/__init__.py @@ -12,9 +12,7 @@ from ._core.queuing.queues.abc import Consumer, Delivery, Producer, Queue from ._core.queuing.queues.memory import MemoryQueue from ._core.related_events import AnyIORelatedEvents, RelatedEvents -from ._core.routing.command_pipeline import ( - ContextCommandPipeline as _ContextCommandPipeline, -) +from ._core.routing.command_pipeline import ContextCommandPipeline from ._core.routing.di import DIAdapter from ._core.routing.dispatchers.abc import Dispatcher from ._core.routing.dispatchers.bus import Bus @@ -77,9 +75,4 @@ new_event_bus = __router__.new_event_bus new_query_bus = __router__.new_query_bus - -class ContextCommandPipeline[C: Command](_ContextCommandPipeline[C]): - __slots__ = () - - def __init__(self, di: DIAdapter = __router__.di) -> None: - super().__init__(di) +ContextCommandPipeline._set_default_di(__router__.di) diff --git a/cq/_core/routing/command_pipeline.py b/cq/_core/routing/command_pipeline.py index a105d6d..526de4b 100644 --- a/cq/_core/routing/command_pipeline.py +++ b/cq/_core/routing/command_pipeline.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Any, Self, overload +from typing import TYPE_CHECKING, Any, ClassVar, Self, overload from cq._core.common.typing import Decorator from cq._core.message import Command, CommandBus, Query, QueryBus @@ -18,7 +18,10 @@ class ContextCommandPipeline[C: Command](ContextPipeline[C]): __query_dispatcher: Dispatcher[Query, Any] - def __init__(self, di: DIAdapter) -> None: + __default_di: ClassVar[DIAdapter] + + def __init__(self, di: DIAdapter | None = None, /) -> None: + di = di or self.__default_di super().__init__(LazyDispatcher(CommandBus, di)) self.__query_dispatcher = LazyDispatcher(QueryBus, di) @@ -50,3 +53,7 @@ def query_step[Q: Query]( # type: ignore[misc] /, ) -> Any: return self.step(wrapped, dispatcher=self.__query_dispatcher) + + @classmethod + def _set_default_di(cls, di: DIAdapter, /) -> None: + cls.__default_di = di diff --git a/cq/_core/routing/router.py b/cq/_core/routing/router.py index fe7cac5..3029ecc 100644 --- a/cq/_core/routing/router.py +++ b/cq/_core/routing/router.py @@ -2,6 +2,7 @@ from typing import Any, Self from cq._core.message import Command, Event, Query +from cq._core.routing.command_pipeline import ContextCommandPipeline from cq._core.routing.di import DIAdapter, NoDI from cq._core.routing.dispatchers.bus import Bus, SimpleBus, TaskBus from cq._core.routing.handler import ( @@ -54,6 +55,9 @@ def query_handler(self) -> HandlerDecorator[Query, Any]: def query_types(self) -> KeysView[type[Query]]: return self.__query_registry.message_types + def command_pipeline[T](self) -> ContextCommandPipeline[T]: + return ContextCommandPipeline(self.__di) + def new_command_bus(self) -> Bus[Command, Any]: bus = SimpleBus(self.__command_registry) diff --git a/tests/test_context_command_pipeline.py b/tests/test_context_command_pipeline.py index 36091ba..1957bfe 100644 --- a/tests/test_context_command_pipeline.py +++ b/tests/test_context_command_pipeline.py @@ -47,9 +47,7 @@ class Context: bar: Bar baz: Baz - pipeline: ContextCommandPipeline[Command0] = ContextCommandPipeline( - router.di - ) + pipeline: ContextCommandPipeline[Command0] = router.command_pipeline() pipeline.add_static_step(Command1())