feat: ✨ Migrate asyncio to anyio

Author: remimd

.github/workflows/cd.yml

@@ -8,6 +8,8 @@ jobs:
   cd:
     name: Continuous Delivery
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
 
     steps:
       - name: Run checkout

.github/workflows/ci.yml

@@ -13,8 +13,10 @@ jobs:
       matrix:
         python-version: ["3.12", "3.13"]
 
-    name: Python ${{ matrix.python-version }}
+    name: Continuous Integration ・ Python ${{ matrix.python-version }}
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
 
     steps:
       - name: Run checkout

conftest.py

@@ -1,9 +1,12 @@
+from collections.abc import Iterator
 from typing import Any
 
 import pytest
+from injection.testing import load_test_profile, set_test_constant
 
-from cq import Bus
+from cq import Bus, CommandBus, EventBus, QueryBus
 from cq._core.dispatcher.bus import SimpleBus
+from cq._core.message import new_command_bus, new_event_bus, new_query_bus
 from tests.helpers.history import HistoryMiddleware
 
 
@@ -15,3 +18,17 @@ def bus() -> Bus[Any, Any]:
 @pytest.fixture(scope="function")
 def history() -> HistoryMiddleware:
     return HistoryMiddleware()
+
+
+@pytest.fixture(scope="function", autouse=True)
+def ensure_test_dependencies(history: HistoryMiddleware) -> Iterator[None]:
+    command_bus: CommandBus[Any] = new_command_bus().add_middlewares(history)
+    event_bus: EventBus = new_event_bus().add_middlewares(history)
+    query_bus: QueryBus[Any] = new_query_bus().add_middlewares(history)
+
+    set_test_constant(command_bus, on=CommandBus, alias=True, mode="override")
+    set_test_constant(event_bus, on=EventBus, alias=True, mode="override")
+    set_test_constant(query_bus, on=QueryBus, alias=True, mode="override")
+
+    with load_test_profile():
+        yield

cq/__init__.py

@@ -18,10 +18,13 @@
     query_handler,
 )
 from ._core.middleware import Middleware, MiddlewareResult
+from ._core.related_events import RelatedEvents
+from ._core.scope import CQScope
 
 __all__ = (
     "AnyCommandBus",
     "Bus",
+    "CQScope",
     "Command",
     "CommandBus",
     "DTO",
@@ -33,6 +36,7 @@
     "Pipe",
     "Query",
     "QueryBus",
+    "RelatedEvents",
     "command_handler",
     "event_handler",
     "get_command_bus",

cq/_core/dispatcher/base.py

@@ -1,4 +1,3 @@
-import asyncio
 from abc import ABC, abstractmethod
 from collections.abc import Awaitable, Callable
 from typing import Protocol, Self, runtime_checkable
@@ -14,10 +13,6 @@ class Dispatcher[I, O](Protocol):
     async def dispatch(self, input_value: I, /) -> O:
         raise NotImplementedError
 
-    @abstractmethod
-    def dispatch_no_wait(self, *input_values: I) -> None:
-        raise NotImplementedError
-
     @abstractmethod
     def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
         raise NotImplementedError
@@ -31,12 +26,6 @@ class BaseDispatcher[I, O](Dispatcher[I, O], ABC):
     def __init__(self) -> None:
         self.__middleware_group = MiddlewareGroup()
 
-    def dispatch_no_wait(self, *input_values: I) -> None:
-        asyncio.gather(
-            *(self.dispatch(input_value) for input_value in input_values),
-            return_exceptions=True,
-        )
-
     def add_middlewares(self, *middlewares: Middleware[[I], O]) -> Self:
         self.__middleware_group.add(*middlewares)
         return self

cq/_core/dispatcher/bus.py

@@ -1,12 +1,12 @@
-import asyncio
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
-from inspect import isclass
+from inspect import getmro, isclass
 from types import GenericAlias
 from typing import Any, Protocol, Self, TypeAliasType, runtime_checkable
 
+import anyio
 import injection
 
 from cq._core.dispatcher.base import BaseDispatcher, Dispatcher
@@ -52,18 +52,12 @@ def decorator(wrapped: type[Handler[[I], O]]) -> type[Handler[[I], O]]:
                 raise TypeError(f"`{wrapped}` isn't a valid handler.")
 
             bus = self.injection_module.find_instance(self.bus_type)
-            lazy_instance = self.injection_module.aget_lazy_instance(
-                wrapped,
-                default=NotImplemented,
-            )
-
-            async def getter() -> Handler[[I], O]:
-                return await lazy_instance
+            factory = self.injection_module.make_async_factory(wrapped)
 
             for input_type in (first_input_type, *input_types):
-                bus.subscribe(input_type, getter)
+                bus.subscribe(input_type, factory)
 
-            return self.injection_module.injectable(wrapped)
+            return wrapped
 
         return decorator
 
@@ -82,7 +76,24 @@ def add_listeners(self, *listeners: Listener[I]) -> Self:
         return self
 
     async def _trigger_listeners(self, input_value: I, /) -> None:
-        await asyncio.gather(*(listener(input_value) for listener in self.__listeners))
+        listeners = self.__listeners
+
+        if not listeners:
+            return
+
+        async with anyio.create_task_group() as task_group:
+            for listener in listeners:
+                task_group.start_soon(listener, input_value)
+
+    @staticmethod
+    def _make_handle_function(
+        handler_factory: HandlerFactory[[I], O],
+    ) -> Callable[[I], Awaitable[O]]:
+        async def handle(input_value: I) -> O:
+            handler = await handler_factory()
+            return await handler.handle(input_value)
+
+        return handle
 
 
 class SimpleBus[I, O](BaseBus[I, O]):
@@ -96,15 +107,16 @@ def __init__(self) -> None:
 
     async def dispatch(self, input_value: I, /) -> O:
         await self._trigger_listeners(input_value)
-        input_type = type(input_value)
 
-        try:
-            handler_factory = self.__handlers[input_type]
-        except KeyError:
+        for input_type in getmro(type(input_value)):
+            if handler_factory := self.__handlers.get(input_type):
+                break
+
+        else:
             return NotImplemented
 
-        handler = await handler_factory()
-        return await self._invoke_with_middlewares(handler.handle, input_value)
+        handler = self._make_handle_function(handler_factory)
+        return await self._invoke_with_middlewares(handler, input_value)
 
     def subscribe(self, input_type: type[I], factory: HandlerFactory[[I], O]) -> Self:
         if input_type in self.__handlers:
@@ -127,20 +139,22 @@ def __init__(self) -> None:
 
     async def dispatch(self, input_value: I, /) -> None:
         await self._trigger_listeners(input_value)
-        handler_factories = self.__handlers.get(type(input_value))
 
-        if not handler_factories:
+        for input_type in getmro(type(input_value)):
+            if handler_factories := self.__handlers.get(input_type):
+                break
+
+        else:
             return
 
-        await asyncio.gather(
-            *[
-                self._invoke_with_middlewares(
-                    (await handler_factory()).handle,
+        async with anyio.create_task_group() as task_group:
+            for handler_factory in handler_factories:
+                handler = self._make_handle_function(handler_factory)
+                task_group.start_soon(
+                    self._invoke_with_middlewares,
+                    handler,
                     input_value,
                 )
-                for handler_factory in handler_factories
-            ]
-        )
 
     def subscribe(
         self,

cq/_core/dispatcher/pipe.py

@@ -24,14 +24,14 @@ def __init__(self, dispatcher: Dispatcher[Any, Any]) -> None:
         self.__dispatcher = dispatcher
         self.__steps = []
 
-    def step[T](  # type: ignore[no-untyped-def]
+    def step[T](
         self,
         wrapped: PipeConverter[T, Any] | None = None,
         /,
         *,
         dispatcher: Dispatcher[T, Any] | None = None,
-    ):
-        def decorator(wp):  # type: ignore[no-untyped-def]
+    ) -> Any:
+        def decorator(wp: PipeConverter[T, Any]) -> PipeConverter[T, Any]:
             step = PipeStep(wp, dispatcher)
             self.__steps.append(step)
             return wp

cq/_core/message.py

@@ -5,6 +5,8 @@
 
 from cq._core.dispatcher.bus import Bus, SimpleBus, SubscriberDecorator, TaskBus
 from cq._core.dto import DTO
+from cq._core.scope import CQScope
+from cq.middlewares.scope import InjectionScopeMiddleware
 
 
 class Message(DTO, ABC):
@@ -34,21 +36,38 @@ class Query(Message, ABC):
 event_handler: SubscriberDecorator[Event, None] = SubscriberDecorator(EventBus)
 query_handler: SubscriberDecorator[Query, Any] = SubscriberDecorator(QueryBus)
 
-injection.set_constant(SimpleBus(), CommandBus, alias=True)
-injection.set_constant(TaskBus(), EventBus, alias=True)
-injection.set_constant(SimpleBus(), QueryBus, alias=True)
-
 
 @injection.inject
 def get_command_bus[T](bus: CommandBus[T] = NotImplemented, /) -> CommandBus[T]:
     return bus
 
 
+def new_command_bus[T]() -> CommandBus[T]:
+    bus: CommandBus[T] = SimpleBus()
+    bus.add_middlewares(
+        InjectionScopeMiddleware(CQScope.ON_COMMAND),
+    )
+    return bus
+
+
 @injection.inject
 def get_event_bus(bus: EventBus = NotImplemented, /) -> EventBus:
     return bus
 
 
+def new_event_bus() -> EventBus:
+    return TaskBus()
+
+
 @injection.inject
 def get_query_bus[T](bus: QueryBus[T] = NotImplemented, /) -> QueryBus[T]:
     return bus
+
+
+def new_query_bus[T]() -> QueryBus[T]:
+    return SimpleBus()
+
+
+injection.set_constant(new_command_bus(), CommandBus, alias=True)
+injection.set_constant(new_event_bus(), EventBus, alias=True)
+injection.set_constant(new_query_bus(), QueryBus, alias=True)

cq/_core/related_events.py

@@ -0,0 +1,43 @@
+from abc import abstractmethod
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+from typing import Protocol, runtime_checkable
+
+import anyio
+import injection
+
+from cq._core.message import Event, EventBus
+from cq._core.scope import CQScope
+
+
+@runtime_checkable
+class RelatedEvents(Protocol):
+    __slots__ = ()
+
+    @abstractmethod
+    def add(self, *events: Event) -> None:
+        raise NotImplementedError
+
+
+@dataclass(frozen=True, slots=True)
+class _RelatedEvents(RelatedEvents):
+    items: list[Event] = field(default_factory=list)
+
+    def add(self, *events: Event) -> None:
+        self.items.extend(events)
+
+
+@injection.scoped(CQScope.ON_COMMAND)
+async def _related_events_recipe(event_bus: EventBus) -> AsyncIterator[RelatedEvents]:
+    yield (instance := _RelatedEvents())
+    events = instance.items
+
+    if not events:
+        return
+
+    async with anyio.create_task_group() as task_group:
+        for event in events:
+            task_group.start_soon(event_bus.dispatch, event)
+
+
+del _related_events_recipe

cq/_core/scope.py

@@ -0,0 +1,5 @@
+from enum import StrEnum, auto
+
+
+class CQScope(StrEnum):
+    ON_COMMAND = auto()