AsyncManagerBase

Warning

Experimental. This class is a prototype. The API and behaviour may change in future releases without a deprecation cycle.

A variant of ManagerBase whose event loop is driven by asyncio instead of the blocking multiprocessing.connection.wait() loop.

---

Difference from synchronous ManagerBase

Aspect	ManagerBase	AsyncManagerBase
Event loop	Blocking wait() loop	asyncio.run()
Handler type	def only	def or async def
after_start()	Before the loop starts	Inside the running event loop; loop-bound clients and pools created there belong to the handlers' loop
Concurrency model	Single-threaded, one event at a time	Async handlers run as concurrent asyncio.Tasks
Worker support	WorkerThread, WorkerProcess	Same (unchanged)
Thread/process pool	Available via Workers	Available via Workers (same as base)

Everything else — @request/@response decorator registration, health checks, workers.start(), after_start(), resolve_callback_worker, resolve_unknown — follows ManagerBase. Cache handling, bookkeeping, response preparation, and Router output use the same shared stages, including the Outgoing-event hook: prepare_outgoing_event().

---

When to use

Use AsyncManagerBase when your Manager handles many concurrent I/O-bound operations and you want to avoid the overhead of a WorkerThread or WorkerProcess for each one. Examples:

• Fetching data from multiple external HTTP APIs simultaneously

• Database queries where the driver supports async (e.g. asyncpg, aiosqlite)

• Waiting on timers, WebSocket messages, or other async-aware I/O

For CPU-bound work, the same rule applies as with synchronous managers: offload to a WorkerProcess via workers.start(). asyncio does not grant extra cores.

---

Async handlers

A @request or @response handler on AsyncManagerBase may be async def:

from edri.abstract import AsyncManagerBase, request


class FetchManager(AsyncManagerBase):
    @request
    async def fetch_url(self, event: FetchUrl) -> None:
        async with aiohttp.ClientSession() as session:
            async with session.get(event.url) as resp:
                event.response.body = await resp.text()

Each async handler runs as its own asyncio.Task. The event loop does not wait for it to finish before reading the next pipe message, so multiple async handlers can run concurrently. Plain def handlers still execute synchronously in the order they are received.

---

Limitations

• No dedicated thread/process pool for handler dispatch. Async handlers run inside the Manager's own asyncio event loop. CPU-bound handlers block the loop just as they would block the sync loop — use WorkerProcess for those.

• Async integration only. The Manager is designed for an async-first application. If most of your handlers are synchronous, the plain ManagerBase is a better fit.

• Experimental. The asyncio event loop and task dispatch remain experimental even though resolver bookkeeping and Router output share the same stages as ManagerBase.

---

Quick example

import asyncio
from edri import EDRI
from edri.abstract import AsyncManagerBase, request
from edri.api.dataclass.api_event import api
from edri.dataclass.event import Event
from edri.dataclass.response import Response, response
from http import HTTPMethod


@response
class PingResponse(Response):
    result: str


@api(url="/ping", method=HTTPMethod.GET)
class Ping(Event):
    response: PingResponse


class PingManager(AsyncManagerBase):
    @request
    async def ping(self, event: Ping) -> None:
        await asyncio.sleep(0.1)
        event.response.result = "pong"


if __name__ == "__main__":
    edri = EDRI()
    edri.add_component(PingManager(edri.router_queue))
    edri.start_api()
    edri.run()

---

See also

• Manager — synchronous ManagerBase and ManagerPriorityBase

• Worker — Worker types for CPU-bound work

• Event — event lifecycle