feat: ✨ Static pipe step

Author: remimd

cq/_core/dispatcher/base.py

@@ -15,7 +15,7 @@ async def dispatch(self, input_value: I, /) -> O:
         raise NotImplementedError
 
     @abstractmethod
-    def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
+    def dispatch_no_wait(self, *input_values: I) -> None:
         raise NotImplementedError
 
     @abstractmethod
@@ -31,12 +31,9 @@ class BaseDispatcher[I, O](Dispatcher[I, O], ABC):
     def __init__(self) -> None:
         self.__middleware_group = MiddlewareGroup()
 
-    def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
+    def dispatch_no_wait(self, *input_values: I) -> None:
         asyncio.gather(
-            *(
-                self.dispatch(input_value)
-                for input_value in (first_input_value, *input_values)
-            ),
+            *(self.dispatch(input_value) for input_value in input_values),
             return_exceptions=True,
         )
 

cq/_core/dispatcher/bus.py

@@ -40,13 +40,8 @@ class SubscriberDecorator[I, O]:
     bus_type: BusType[I, O] | TypeAliasType | GenericAlias
     injection_module: injection.Module = field(default_factory=injection.mod)
 
-    def __call__[T](
-        self,
-        first_input_type: type[I],
-        /,
-        *input_types: type[I],
-    ) -> Callable[[T], T]:
-        def decorator(wrapped: T) -> T:
+    def __call__(self, first_input_type: type[I], /, *input_types: type[I]):  # type: ignore[no-untyped-def]
+        def decorator(wrapped):  # type: ignore[no-untyped-def]
             if not isclass(wrapped) or not issubclass(wrapped, Handler):
                 raise TypeError(f"`{wrapped}` isn't a valid handler.")
 
@@ -56,7 +51,7 @@ def decorator(wrapped: T) -> T:
             for input_type in (first_input_type, *input_types):
                 bus.subscribe(input_type, factory)
 
-            return wrapped  # type: ignore[return-value]
+            return wrapped
 
         return decorator
 

cq/_core/dispatcher/pipe.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from dataclasses import dataclass, field
-from typing import Any, Awaitable
+from typing import Any, Awaitable, Self
 
 from cq._core.dispatcher.base import BaseDispatcher, Dispatcher
 
@@ -38,6 +38,18 @@ def decorator(wp):  # type: ignore[no-untyped-def]
 
         return decorator(wrapped) if wrapped else decorator
 
+    def add_static_step[T](
+        self,
+        input_value: T,
+        *,
+        dispatcher: Dispatcher[T, Any] | None = None,
+    ) -> Self:
+        @self.step(dispatcher=dispatcher)
+        async def converter(_: Any) -> T:
+            return input_value
+
+        return self
+
     async def dispatch(self, input_value: I, /) -> O:
         return await self._invoke_with_middlewares(self.__execute, input_value)
 

cq/_core/middleware.py

@@ -2,6 +2,8 @@
 from dataclasses import dataclass, field
 from typing import Self
 
+from cq.exceptions import MiddlewareError
+
 type MiddlewareResult[T] = AsyncGenerator[None, T]
 type Middleware[**P, T] = Callable[P, MiddlewareResult[T]]
 
@@ -47,7 +49,9 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
                         await generator.athrow(exc)
                     else:
                         await generator.asend(value)
-                        break
+                        raise MiddlewareError(
+                            f"Too many `yield` keywords in `{middleware}`."
+                        )
 
             except StopAsyncIteration:
                 ...

cq/exceptions.py

@@ -0,0 +1,7 @@
+__all__ = ("CQError", "MiddlewareError")
+
+
+class CQError(Exception): ...
+
+
+class MiddlewareError(CQError): ...

cq/middlewares/retry.py

@@ -3,6 +3,8 @@
 
 from cq import MiddlewareResult
 
+__all__ = ("RetryMiddleware",)
+
 
 class RetryMiddleware:
     __slots__ = ("__delay", "__exceptions", "__retry")

tests/core/dispatcher/test_pipe.py

@@ -32,6 +32,7 @@ async def step_converter_1(length: int) -> int:
 
         assert await pipe.dispatch("hello") == "*****"
 
+        # Custom dispatcher
         other_bus: Bus[Any, Any] = SimpleBus()
 
         other_bus.subscribe(str, ToTupleHandler)
@@ -41,3 +42,8 @@ async def step_converter_2(hidden_string: str) -> str:
             return hidden_string
 
         assert await pipe.dispatch("hello") == ("*", "*", "*", "*", "*")
+
+        # Add static step
+        pipe.add_static_step(2)
+
+        assert await pipe.dispatch("hello") == "**"

tests/core/test_middleware.py

@@ -1,10 +1,9 @@
-from __future__ import annotations
-
 from typing import Any
 
 import pytest
 
 from cq._core.middleware import MiddlewareGroup, MiddlewareResult
+from cq.exceptions import MiddlewareError
 from tests.helpers.history import HistoryMiddleware
 
 
@@ -62,6 +61,24 @@ async def handler() -> str:
         assert isinstance(record.result, ValueError)
         assert record.is_failed
 
+    async def test_invoke_with_too_many_yield_raise_middleware_error(
+        self,
+        group: MiddlewareGroup[..., Any],
+    ) -> None:
+        async def handler() -> None: ...
+
+        async def too_many_yield_middleware(
+            *args: Any,
+            **kwargs: Any,
+        ) -> MiddlewareResult[Any]:
+            yield
+            yield
+
+        group.add(too_many_yield_middleware)
+
+        with pytest.raises(MiddlewareError):
+            await group.invoke(handler)
+
     async def test_invoke_with_multiple_yield_return_any(
         self,
         group: MiddlewareGroup[..., Any],