feat: ✨ Retry middleware

Author: remimd

conftest.py

@@ -0,0 +1,17 @@
+from typing import Any
+
+import pytest
+
+from cq import Bus
+from cq._core.bus import SimpleBus
+from tests.helpers.history import HistoryMiddleware
+
+
+@pytest.fixture(scope="function")
+def bus() -> Bus[Any, Any]:
+    return SimpleBus()
+
+
+@pytest.fixture(scope="function")
+def history() -> HistoryMiddleware:
+    return HistoryMiddleware()

cq/__init__.py

@@ -1,10 +1,12 @@
+from ._core.bus import Bus
 from ._core.command import Command, CommandBus, command_handler, find_command_bus
 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__ = (
+    "Bus",
     "Command",
     "CommandBus",
     "DTO",

cq/_core/bus.py

@@ -39,7 +39,8 @@ def dispatch_no_wait(self, first_input_value: I, /, *input_values: I) -> None:
             *(
                 self.dispatch(input_value)
                 for input_value in (first_input_value, *input_values)
-            )
+            ),
+            return_exceptions=True,
         )
 
     @abstractmethod

cq/middlewares/__init__.py

@@ -0,0 +1,34 @@
+import asyncio
+from typing import Any
+
+from cq import MiddlewareResult
+
+
+class RetryMiddleware:
+    __slots__ = ("__delay", "__exceptions", "__retry")
+
+    def __init__(
+        self,
+        retry: int,
+        delay: float = 0,
+        exceptions: tuple[type[BaseException], ...] = (Exception,),
+    ) -> None:
+        self.__delay = delay
+        self.__exceptions = exceptions
+        self.__retry = retry
+
+    async def __call__(self, *args: Any, **kwargs: Any) -> MiddlewareResult[Any]:
+        retry = self.__retry
+
+        for attempt in range(1, retry + 1):
+            try:
+                yield
+
+            except self.__exceptions as exc:
+                if attempt == retry:
+                    raise exc
+
+            else:
+                break
+
+            await asyncio.sleep(self.__delay)

cq/middlewares/retry.py

@@ -9,10 +9,6 @@
 
 
 class TestSimpleBus:
-    @pytest.fixture(scope="function")
-    def bus(self) -> SimpleBus[Any, Any]:
-        return SimpleBus()
-
     def test_add_middlewares_with_success_return_self(
         self,
         bus: SimpleBus[Any, Any],
@@ -93,26 +89,29 @@ class BadHandler: ...
 
 class TestTaskBus:
     @pytest.fixture(scope="function")
-    def bus(self) -> TaskBus[Any]:
+    def task_bus(self) -> TaskBus[Any]:
         return TaskBus()
 
-    def test_subscribe_with_success_return_self(self, bus: TaskBus[Any]) -> None:
-        assert bus.subscribe(str, _SomeTaskHandler) is bus
+    def test_subscribe_with_success_return_self(self, task_bus: TaskBus[Any]) -> None:
+        assert task_bus.subscribe(str, _SomeTaskHandler) is task_bus
         # Checks whether several handlers can be subscribed for the same input type
-        assert bus.subscribe(str, _SomeTaskHandler) is bus
+        assert task_bus.subscribe(str, _SomeTaskHandler) is task_bus
 
-    async def test_dispatch_with_success_return_none(self, bus: TaskBus[Any]) -> None:
-        bus.subscribe(str, _SomeTaskHandler)
+    async def test_dispatch_with_success_return_none(
+        self,
+        task_bus: TaskBus[Any],
+    ) -> None:
+        task_bus.subscribe(str, _SomeTaskHandler)
 
         with pytest.raises(NotImplementedError):
-            await bus.dispatch("hello")
+            await task_bus.dispatch("hello")
 
     async def test_dispatch_with_unknown_input_type_return_none(
         self,
-        bus: TaskBus[Any],
+        task_bus: TaskBus[Any],
     ) -> None:
         # Do nothing
-        await bus.dispatch("hello")
+        await task_bus.dispatch("hello")
 
 
 class _SomeHandler:

poetry.lock

@@ -1,34 +1,29 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
-from enum import IntEnum
-from typing import Any, NamedTuple
+from typing import Any
 
 import pytest
 
 from cq._core.middleware import MiddlewareGroup, MiddlewareResult
+from tests.helpers.history import HistoryMiddleware
 
 
 class TestMiddlewareGroup:
     @pytest.fixture(scope="function")
     def group(self) -> MiddlewareGroup[..., Any]:
         return MiddlewareGroup()
 
-    @pytest.fixture(scope="function")
-    def history(self) -> _HistoryMiddleware:
-        return _HistoryMiddleware()
-
     def test_add_with_success_return_self(
         self,
         group: MiddlewareGroup[..., Any],
-        history: _HistoryMiddleware,
+        history: HistoryMiddleware,
     ) -> None:
         assert group.add(history) is group
 
     async def test_invoke_with_success_return_any(
         self,
         group: MiddlewareGroup[..., Any],
-        history: _HistoryMiddleware,
+        history: HistoryMiddleware,
     ) -> None:
         async def handler() -> str:
             return "I'm a handler..."
@@ -43,18 +38,20 @@ async def handler() -> str:
         assert record.args == ()
         assert record.kwargs == {}
         assert record.result == result
-        assert record.status == _Status.SUCCESS
+        assert record.is_success
 
     async def test_invoke_with_exception_raise_any(
         self,
         group: MiddlewareGroup[..., Any],
-        history: _HistoryMiddleware,
+        history: HistoryMiddleware,
     ) -> None:
         async def handler() -> str:
             raise ValueError("I failed...")
 
         group.add(history)
-        assert await group.invoke(handler) is NotImplemented
+
+        with pytest.raises(ValueError):
+            await group.invoke(handler)
 
         records = history.records
         assert len(records) == 1
@@ -63,12 +60,12 @@ async def handler() -> str:
         assert record.args == ()
         assert record.kwargs == {}
         assert isinstance(record.result, ValueError)
-        assert record.status == _Status.FAILED
+        assert record.is_failed
 
     async def test_invoke_with_multiple_yield_return_any(
         self,
         group: MiddlewareGroup[..., Any],
-        history: _HistoryMiddleware,
+        history: HistoryMiddleware,
     ) -> None:
         async def handler() -> str:
             return "I'm a handler..."
@@ -80,37 +77,6 @@ async def handler() -> str:
         assert len(records) == 2
 
 
-class _Status(IntEnum):
-    SUCCESS = 1
-    FAILED = 0
-
-
-class _Record(NamedTuple):
-    args: tuple[Any, ...]
-    kwargs: Mapping[str, Any]
-    result: Any
-    status: _Status
-
-
-class _HistoryMiddleware:
-    def __init__(self) -> None:
-        self.__records: list[_Record] = []
-
-    @property
-    def records(self) -> tuple[_Record, ...]:
-        return tuple(self.__records)
-
-    async def __call__(self, /, *args: Any, **kwargs: Any) -> MiddlewareResult[Any]:
-        try:
-            result = yield
-        except BaseException as exc:
-            record = _Record(args, kwargs, exc, _Status.FAILED)
-        else:
-            record = _Record(args, kwargs, result, _Status.SUCCESS)
-
-        self.__records.append(record)
-
-
 async def _exec_2_times_middleware(*args: Any, **kwargs: Any) -> MiddlewareResult[Any]:
     yield
     yield

tests/core/test_bus.py

@@ -0,0 +1,44 @@
+from enum import IntEnum
+from typing import Any, Mapping, NamedTuple
+
+from cq import MiddlewareResult
+
+
+class HistoryRecordStatus(IntEnum):
+    SUCCESS = 1
+    FAILED = 0
+
+
+class HistoryRecord(NamedTuple):
+    args: tuple[Any, ...]
+    kwargs: Mapping[str, Any]
+    result: Any
+    status: HistoryRecordStatus
+
+    @property
+    def is_success(self) -> bool:
+        return self.status == HistoryRecordStatus.SUCCESS
+
+    @property
+    def is_failed(self) -> bool:
+        return self.status == HistoryRecordStatus.FAILED
+
+
+class HistoryMiddleware:
+    def __init__(self) -> None:
+        self.__records: list[HistoryRecord] = []
+
+    @property
+    def records(self) -> tuple[HistoryRecord, ...]:
+        return tuple(self.__records)
+
+    async def __call__(self, /, *args: Any, **kwargs: Any) -> MiddlewareResult[Any]:
+        try:
+            result = yield
+        except BaseException as exc:
+            record = HistoryRecord(args, kwargs, exc, HistoryRecordStatus.FAILED)
+            raise exc
+        else:
+            record = HistoryRecord(args, kwargs, result, HistoryRecordStatus.SUCCESS)
+        finally:
+            self.__records.append(record)