feat: ✨ Pipeline

Author: remimd

README.md

@@ -21,4 +21,5 @@ pip install python-cq
 ## Resources
 
 * [**Writing Application Layer**](https://github.com/100nm/python-cq/tree/prod/documentation/writing-application-layer.md)
+* [**Pipeline**](https://github.com/100nm/python-cq/tree/prod/documentation/pipeline.md)
 * [**FastAPI Example**](https://github.com/100nm/python-cq/tree/prod/documentation/fastapi-example.md)

conftest.py

@@ -3,7 +3,7 @@
 import pytest
 
 from cq import Bus
-from cq._core.bus import SimpleBus
+from cq._core.dispatcher.bus import SimpleBus
 from tests.helpers.history import HistoryMiddleware
 
 

cq/__init__.py

@@ -1,11 +1,19 @@
-from ._core.bus import Bus
-from ._core.command import Command, CommandBus, command_handler, find_command_bus
+from ._core.command import (
+    AnyCommandBus,
+    Command,
+    CommandBus,
+    command_handler,
+    find_command_bus,
+)
+from ._core.dispatcher.bus import Bus
+from ._core.dispatcher.pipe import Pipe
 from ._core.dto import DTO
 from ._core.event import Event, EventBus, event_handler, find_event_bus
 from ._core.middleware import Middleware, MiddlewareResult
 from ._core.query import Query, QueryBus, find_query_bus, query_handler
 
 __all__ = (
+    "AnyCommandBus",
     "Bus",
     "Command",
     "CommandBus",
@@ -14,6 +22,7 @@
     "EventBus",
     "Middleware",
     "MiddlewareResult",
+    "Pipe",
     "Query",
     "QueryBus",
     "command_handler",

cq/_core/command.py

@@ -3,7 +3,7 @@
 
 import injection
 
-from cq._core.bus import Bus, SimpleBus, SubscriberDecorator
+from cq._core.dispatcher.bus import Bus, SimpleBus, SubscriberDecorator
 from cq._core.dto import DTO
 
 
@@ -12,6 +12,7 @@ class Command(DTO, ABC):
 
 
 type CommandBus[T] = Bus[Command, T]
+AnyCommandBus = CommandBus[Any]
 command_handler: SubscriberDecorator[Command, Any] = SubscriberDecorator(CommandBus)
 
 injection.set_constant(SimpleBus(), CommandBus, alias=True)

cq/_core/dispatcher/__init__.py

@@ -0,0 +1,53 @@
+import asyncio
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable
+from typing import Callable, Protocol, Self, runtime_checkable
+
+from cq._core.middleware import Middleware, MiddlewareGroup
+
+
+@runtime_checkable
+class Dispatcher[I, O](Protocol):
+    __slots__ = ()
+
+    @abstractmethod
+    async def dispatch(self, input_value: I, /) -> O:
+        raise NotImplementedError
+
+    @abstractmethod
+    def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
+        raise NotImplementedError
+
+
+class BaseDispatcher[I, O](Dispatcher[I, O], ABC):
+    __slots__ = ("__middleware_group",)
+
+    __middleware_group: MiddlewareGroup[[I], O]
+
+    def __init__(self) -> None:
+        self.__middleware_group = MiddlewareGroup()
+
+    def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
+        asyncio.gather(
+            *(
+                self.dispatch(input_value)
+                for input_value in (first_input_value, *input_values)
+            ),
+            return_exceptions=True,
+        )
+
+    def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
+        self.__middleware_group.add(*middlewares)
+        return self
+
+    async def _invoke_with_middlewares(
+        self,
+        handler: Callable[[I], Awaitable[O]],
+        input_value: I,
+        /,
+    ) -> O:
+        return await self.__middleware_group.invoke(handler, input_value)

cq/_core/dispatcher/base.py

@@ -1,5 +1,5 @@
 import asyncio
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from collections import defaultdict
 from collections.abc import Callable
 from dataclasses import dataclass, field
@@ -9,7 +9,7 @@
 
 import injection
 
-from cq._core.middleware import Middleware, MiddlewareGroup
+from cq._core.dispatcher.base import BaseDispatcher, Dispatcher
 
 type HandlerType[**P, T] = type[Handler[P, T]]
 type HandlerFactory[**P, T] = Callable[..., Handler[P, T]]
@@ -27,30 +27,13 @@ async def handle(self, *args: P.args, **kwargs: P.kwargs) -> T:
 
 
 @runtime_checkable
-class Bus[I, O](Protocol):
+class Bus[I, O](Dispatcher[I, O], Protocol):
     __slots__ = ()
 
-    @abstractmethod
-    async def dispatch(self, input_value: I, /) -> O:
-        raise NotImplementedError
-
-    def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
-        asyncio.gather(
-            *(
-                self.dispatch(input_value)
-                for input_value in (first_input_value, *input_values)
-            ),
-            return_exceptions=True,
-        )
-
     @abstractmethod
     def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
         raise NotImplementedError
 
-    @abstractmethod
-    def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
-        raise NotImplementedError
-
 
 @dataclass(eq=False, frozen=True, slots=True)
 class SubscriberDecorator[I, O]:
@@ -81,23 +64,7 @@ def __find_bus(self) -> Bus[I, O]:
         return self.injection_module.find_instance(self.bus_type)
 
 
-class _BaseBus[I, O](Bus[I, O], ABC):
-    __slots__ = ("__middleware_group",)
-
-    __middleware_group: MiddlewareGroup[[I], O]
-
-    def __init__(self) -> None:
-        self.__middleware_group = MiddlewareGroup()
-
-    def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
-        self.__middleware_group.add(*middlewares)
-        return self
-
-    async def _invoke(self, handler: Handler[[I], O], input_value: I, /) -> O:
-        return await self.__middleware_group.invoke(handler.handle, input_value)
-
-
-class SimpleBus[I, O](_BaseBus[I, O]):
+class SimpleBus[I, O](BaseDispatcher[I, O], Bus[I, O]):
     __slots__ = ("__handlers",)
 
     __handlers: dict[type[I], HandlerFactory[[I], O]]
@@ -114,7 +81,10 @@ async def dispatch(self, input_value: I, /) -> O:
         except KeyError:
             return NotImplemented
 
-        return await self._invoke(handler_factory(), input_value)
+        return await self._invoke_with_middlewares(
+            handler_factory().handle,
+            input_value,
+        )
 
     def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
         if input_type in self.__handlers:
@@ -126,7 +96,7 @@ def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Sel
         return self
 
 
-class TaskBus[I](_BaseBus[I, None]):
+class TaskBus[I](BaseDispatcher[I, None], Bus[I, None]):
     __slots__ = ("__handlers",)
 
     __handlers: dict[type[I], list[HandlerFactory[[I], None]]]
@@ -143,7 +113,10 @@ async def dispatch(self, input_value: I, /) -> None:
 
         await asyncio.gather(
             *(
-                self._invoke(handler_factory(), input_value)
+                self._invoke_with_middlewares(
+                    handler_factory().handle,
+                    input_value,
+                )
                 for handler_factory in handler_factories
             )
         )

cq/_core/bus.py cq/_core/dispatcher/bus.pycq/_core/bus.py renamed to cq/_core/dispatcher/bus.py

@@ -0,0 +1,52 @@
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any, Awaitable
+
+from cq._core.dispatcher.base import BaseDispatcher, Dispatcher
+
+type PipeConverter[I, O] = Callable[[O], Awaitable[I]]
+
+
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
+class PipeStep[I, O]:
+    converter: PipeConverter[I, O]
+    dispatcher: Dispatcher[I, Any] | None = field(default=None)
+
+
+class Pipe[I, O](BaseDispatcher[I, O]):
+    __slots__ = ("__dispatcher", "__steps")
+
+    __dispatcher: Dispatcher[Any, Any]
+    __steps: list[PipeStep[Any, Any]]
+
+    def __init__(self, dispatcher: Dispatcher[Any, Any]) -> None:
+        super().__init__()
+        self.__dispatcher = dispatcher
+        self.__steps = []
+
+    def step[T](  # type: ignore[no-untyped-def]
+        self,
+        wrapped: PipeConverter[T, Any] | None = None,
+        /,
+        *,
+        dispatcher: Dispatcher[T, Any] | None = None,
+    ):
+        def decorator(wp):  # type: ignore[no-untyped-def]
+            step = PipeStep(wp, dispatcher)
+            self.__steps.append(step)
+            return wp
+
+        return decorator(wrapped) if wrapped else decorator
+
+    async def dispatch(self, input_value: I, /) -> O:
+        return await self._invoke_with_middlewares(self.__execute, input_value)
+
+    async def __execute(self, input_value: I) -> O:
+        dispatcher = self.__dispatcher
+
+        for step in self.__steps:
+            output_value = await dispatcher.dispatch(input_value)
+            input_value = await step.converter(output_value)
+            dispatcher = step.dispatcher or self.__dispatcher
+
+        return await dispatcher.dispatch(input_value)

cq/_core/dispatcher/pipe.py

@@ -2,7 +2,7 @@
 
 import injection
 
-from cq._core.bus import Bus, SubscriberDecorator, TaskBus
+from cq._core.dispatcher.bus import Bus, SubscriberDecorator, TaskBus
 from cq._core.dto import DTO
 
 

cq/_core/event.py

@@ -6,13 +6,9 @@
 type Middleware[**P, T] = Callable[P, MiddlewareResult[T]]
 
 
-@dataclass(eq=False, frozen=True, slots=True)
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
 class MiddlewareGroup[**P, T]:
-    __middlewares: list[Middleware[P, T]] = field(
-        default_factory=list,
-        init=False,
-        repr=False,
-    )
+    __middlewares: list[Middleware[P, T]] = field(default_factory=list, init=False)
 
     @property
     def __stack(self) -> Iterator[Middleware[P, T]]:
@@ -51,6 +47,7 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
                         await generator.athrow(exc)
                     else:
                         await generator.asend(value)
+                        break
 
             except StopAsyncIteration:
                 ...