feat: ✨ Introduce DeferredBus

Author: remimd

cq/__init__.py

@@ -1,3 +1,4 @@
+from ._core.defer import DeferredBus
 from ._core.dispatcher.bus import Bus
 from ._core.dispatcher.pipe import Pipe
 from ._core.message import (
@@ -25,6 +26,7 @@
     "CQScope",
     "Command",
     "CommandBus",
+    "DeferredBus",
     "Event",
     "EventBus",
     "Middleware",

cq/_core/defer.py

@@ -0,0 +1,10 @@
+from abc import abstractmethod
+from typing import Protocol
+
+
+class DeferredBus[I](Protocol):
+    __slots__ = ()
+
+    @abstractmethod
+    async def defer(self, input_value: I, /) -> None:
+        raise NotImplementedError

cq/ext/__init__.py

@@ -0,0 +1,44 @@
+from dataclasses import dataclass
+from typing import Annotated, Any
+
+from fastapi import BackgroundTasks, Depends
+from injection.ext.fastapi import Inject
+
+from cq import Bus, Command, CommandBus, DeferredBus, Event, EventBus, Query, QueryBus
+
+__all__ = ("DeferredCommandBus", "DeferredEventBus", "DeferredQueryBus")
+
+
+@dataclass(repr=False, eq=False, frozen=True, slots=True)
+class FastAPIDeferredBus[I](DeferredBus[I]):
+    background_tasks: BackgroundTasks
+    bus: Bus[I, Any]
+
+    async def defer(self, input_value: I, /) -> None:
+        self.background_tasks.add_task(self.bus.dispatch, input_value)
+
+
+async def new_deferred_command_bus[T](
+    background_tasks: BackgroundTasks,
+    command_bus: Inject[CommandBus[T]],
+) -> DeferredBus[Command]:
+    return FastAPIDeferredBus(background_tasks, command_bus)
+
+
+async def new_deferred_event_bus(
+    background_tasks: BackgroundTasks,
+    event_bus: Inject[EventBus],
+) -> DeferredBus[Event]:
+    return FastAPIDeferredBus(background_tasks, event_bus)
+
+
+async def new_deferred_query_bus[T](
+    background_tasks: BackgroundTasks,
+    query_bus: Inject[QueryBus[T]],
+) -> DeferredBus[Query]:
+    return FastAPIDeferredBus(background_tasks, query_bus)
+
+
+DeferredCommandBus = Annotated[DeferredBus[Command], Depends(new_deferred_command_bus)]
+DeferredEventBus = Annotated[DeferredBus[Event], Depends(new_deferred_event_bus)]
+DeferredQueryBus = Annotated[DeferredBus[Query], Depends(new_deferred_query_bus)]

cq/ext/fastapi.py

@@ -0,0 +1,5 @@
+from cq import Command, DeferredBus, Event, Query
+
+type DeferredCommandBus = DeferredBus[Command]
+type DeferredEventBus = DeferredBus[Event]
+type DeferredQueryBus = DeferredBus[Query]

cq/ext/fastapi.pyi

@@ -6,6 +6,7 @@ Here's an example of its integration:
 
 ```python
 from cq import CommandBus, command_handler, new_command_bus
+from cq.ext.fastapi import DeferredCommandBus
 from fastapi import FastAPI, status
 from injection import injectable
 from injection.ext.fastapi import Inject
@@ -48,4 +49,14 @@ async def example(
 ) -> None:
     result = await command_bus.dispatch(command)
     # ...
+
+@app.post("/background-example", status_code=status.HTTP_204_NO_CONTENT)
+async def background_example(
+    command: ExampleCommand,
+    command_bus: DeferredCommandBus,
+) -> None:
+    # runs the command in the background
+    # so the client receives a response more quickly
+    # but isn't notified in case of error
+    await command_bus.defer(command)
 ```

documentation/fastapi-example.md

@@ -14,6 +14,8 @@ example = [
     "pydantic",
 ]
 test = [
+    "fastapi",
+    "httpx",
     "pytest",
     "pytest-asyncio",
     "pytest-cov",

pyproject.toml

@@ -0,0 +1,40 @@
+from collections.abc import Iterator
+from typing import ClassVar
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from cq import command_handler
+from cq.ext.fastapi import DeferredCommandBus
+
+app = FastAPI()
+
+
+class DeferredCommand: ...
+
+
+@command_handler
+class DeferredCommandHandler:
+    call_count: ClassVar[int] = 0
+
+    async def handle(self, command: DeferredCommand) -> None:
+        DeferredCommandHandler.call_count += 1
+
+
+@app.post("/defer", status_code=204)
+async def defer(bus: DeferredCommandBus) -> None:
+    command = DeferredCommand()
+    await bus.defer(command)
+
+
+class TestFastAPIDeferredBus:
+    @pytest.fixture(scope="class")
+    def client(self) -> Iterator[TestClient]:
+        with TestClient(app) as client:
+            yield client
+
+    def test_defer_with_deferred_command_bus(self, client: TestClient) -> None:
+        response = client.post("/defer")
+        assert response.status_code == 204
+        assert DeferredCommandHandler.call_count == 1