refactoring: ♻️ Bus

Author: remimd

cq/_core/dispatcher/bus.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterator
 from typing import Any, Protocol, Self, runtime_checkable
 
 import anyio
@@ -30,69 +30,64 @@ def add_listeners(self, *listeners: Listener[I]) -> Self:
 
 
 class BaseBus[I, O](BaseDispatcher[I, O], Bus[I, O], ABC):
-    __slots__ = ("__listeners",)
+    __slots__ = ("__listeners", "__manager")
 
     __listeners: list[Listener[I]]
+    __manager: HandlerManager[I, O]
 
-    def __init__(self) -> None:
+    def __init__(self, manager: HandlerManager[I, O]) -> None:
         super().__init__()
         self.__listeners = []
+        self.__manager = manager
 
     def add_listeners(self, *listeners: Listener[I]) -> Self:
         self.__listeners.extend(listeners)
         return self
 
+    def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
+        self.__manager.subscribe(input_type, factory)
+        return self
+
+    def _handlers_from(
+        self,
+        input_type: type[I],
+    ) -> Iterator[Callable[[I], Awaitable[O]]]:
+        return self.__manager.handlers_from(input_type)
+
     def _trigger_listeners(self, input_value: I, /, task_group: TaskGroup) -> None:
         for listener in self.__listeners:
             task_group.start_soon(listener, input_value)
 
 
 class SimpleBus[I, O](BaseBus[I, O]):
-    __slots__ = ("__manager",)
-
-    __manager: HandlerManager[I, O]
+    __slots__ = ()
 
     def __init__(self, manager: HandlerManager[I, O] | None = None) -> None:
-        super().__init__()
-        self.__manager = manager or SingleHandlerManager()
+        super().__init__(manager or SingleHandlerManager())
 
     async def dispatch(self, input_value: I, /) -> O:
         async with anyio.create_task_group() as task_group:
             self._trigger_listeners(input_value, task_group)
 
-        for handler in self.__manager.handlers_from(type(input_value)):
+        for handler in self._handlers_from(type(input_value)):
             return await self._invoke_with_middlewares(handler, input_value)
 
         return NotImplemented
 
-    def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
-        self.__manager.subscribe(input_type, factory)
-        return self
-
 
 class TaskBus[I](BaseBus[I, None]):
-    __slots__ = ("__manager",)
-
-    __manager: HandlerManager[I, None]
+    __slots__ = ()
 
     def __init__(self, manager: HandlerManager[I, None] | None = None) -> None:
-        super().__init__()
-        self.__manager = manager or MultipleHandlerManager()
+        super().__init__(manager or MultipleHandlerManager())
 
     async def dispatch(self, input_value: I, /) -> None:
         async with anyio.create_task_group() as task_group:
             self._trigger_listeners(input_value, task_group)
-            for handler in self.__manager.handlers_from(type(input_value)):
+
+            for handler in self._handlers_from(type(input_value)):
                 task_group.start_soon(
                     self._invoke_with_middlewares,
                     handler,
                     input_value,
                 )
-
-    def subscribe(
-        self,
-        input_type: type[I],
-        factory: HandlerFactory[[I], None],
-    ) -> Self:
-        self.__manager.subscribe(input_type, factory)
-        return self

cq/_core/handler.py

@@ -37,7 +37,7 @@ def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Sel
         raise NotImplementedError
 
 
-@dataclass(eq=False, frozen=True, slots=True)
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
 class MultipleHandlerManager[I, O](HandlerManager[I, O]):
     __factories: dict[type[I], list[HandlerFactory[[I], O]]] = field(
         default_factory=partial(defaultdict, list),
@@ -57,7 +57,7 @@ def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Sel
         return self
 
 
-@dataclass(eq=False, frozen=True, slots=True)
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
 class SingleHandlerManager[I, O](HandlerManager[I, O]):
     __factories: dict[type[I], HandlerFactory[[I], O]] = field(
         default_factory=dict,
@@ -83,7 +83,7 @@ def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Sel
         return self
 
 
-@dataclass(eq=False, frozen=True, slots=True)
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
 class HandlerDecorator[I, O]:
     manager: HandlerManager[I, O]
     injection_module: injection.Module = field(default_factory=injection.mod)
@@ -101,9 +101,9 @@ def decorator(wrapped: type[Handler[[I], O]]) -> type[Handler[[I], O]]:
 
 
 def _make_handle_function[I, O](
-    handler_factory: HandlerFactory[[I], O],
+    factory: HandlerFactory[[I], O],
 ) -> Callable[[I], Awaitable[O]]:
-    return partial(__handle, factory=handler_factory)
+    return partial(__handle, factory=factory)
 
 
 async def __handle[I, O](input_value: I, factory: HandlerFactory[[I], O]) -> O:

cq/_core/message.py

@@ -55,9 +55,9 @@ class Query(Message, ABC):
     mode="fallback",
 )
 def new_command_bus[T]() -> CommandBus[T]:
-    return SimpleBus(command_handler.manager).add_middlewares(
-        InjectionScopeMiddleware(CQScope.ON_COMMAND),
-    )
+    bus = SimpleBus(command_handler.manager)
+    bus.add_middlewares(InjectionScopeMiddleware(CQScope.ON_COMMAND))
+    return bus
 
 
 @injection.singleton(

cq/_core/related_events.py

@@ -19,7 +19,7 @@ def add(self, *events: Event) -> None:
         raise NotImplementedError
 
 
-@dataclass(frozen=True, slots=True)
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
 class _RelatedEvents(RelatedEvents):
     items: list[Event] = field(default_factory=list)
 

cq/middlewares/retry.py

@@ -11,6 +11,10 @@
 class RetryMiddleware:
     __slots__ = ("__delay", "__exceptions", "__retry")
 
+    __delay: float
+    __exceptions: tuple[type[BaseException], ...]
+    __retry: int
+
     def __init__(
         self,
         retry: int,

cq/middlewares/scope.py

@@ -13,6 +13,8 @@
 class InjectionScopeMiddleware:
     __slots__ = ("__scope_name",)
 
+    __scope_name: str
+
     def __init__(self, scope_name: str) -> None:
         self.__scope_name = scope_name
 

documentation/fastapi-example.md

@@ -1,7 +1,11 @@
 # [FastAPI](https://github.com/fastapi/fastapi) Example
 
+The advantage of `python-cq` is that it can be easily integrated into FastAPI.
+
+Here's an example of its integration:
+
 ```python
-from cq import Command, CommandBus, command_handler, new_command_bus
+from cq import Command, CommandBus, DTO, command_handler, new_command_bus
 from fastapi import FastAPI, status
 from injection import injectable, singleton
 from injection.integrations.fastapi import Inject
@@ -21,7 +25,7 @@ def override_command_bus_recipe() -> CommandBus:
 
 class ExampleCommand(Command): ...
 
-type HandlerReturnType = ...
+class HandlerReturnType(DTO): ...
 
 @command_handler(ExampleCommand)
 class ExampleHandler:

documentation/pipeline.md

@@ -1,5 +1,7 @@
 # Pipeline
 
+Pipelines are designed to execute several commands one after the other, while encapsulating them in middleware.
+
 Example:
 
 ```python

documentation/writing-application-layer.md

@@ -189,7 +189,7 @@ class SomeMiddleware:
 
 ### Add middleware
 
-To add a middleware: you need to override the bus recipe.
+To add a middleware, you need to override the bus recipe.
 
 ```python
 from cq import CommandBus, new_command_bus