mediary

Typed, decorator-driven mediator + CQRS for Python — handlers, pipelines and notifications discovered by package scan.

• Decorate, don't register. Mark requests with @request and handlers with @handler; one mediator.scan("app") wires the whole package.
• Typed end to end. await mediator.send(GetUser(1)) is typed as User; handlers are plain classes or functions, matched structurally.
• Pipelines. Behaviors (middleware) wrap handlers, targeted by type, Protocol or kind, and ordered. Logging, retry and timeout ship ready-made.
• Notifications, with sequential or concurrent publishing.
• CQRS pack. @command, @query, @event, and senders that can only send one kind.
• Pluggable DI, testing helpers and a pytest fixture. Zero dependencies, asyncio only, Python 3.11+.

Install

pip install mediary      # or: uv add mediary

Quickstart

Declare a request, what it returns, and its handler:

from dataclasses import dataclass

from mediary import Returns, handler, request


@request
@dataclass
class PlaceOrder(Returns[int]):
    item: str
    quantity: int


@handler
class PlaceOrderHandler:
    async def handle(self, request: PlaceOrder) -> int:
        return 42  # the new order's id

Scan the package once at startup, then send requests from anywhere:

from mediary import Mediator
from shop.orders import PlaceOrder

mediator = Mediator()
mediator.scan("shop")  # imports shop and its submodules, registers each @handler

order_id = await mediator.send(PlaceOrder("book", 2))  # typed as int
assert order_id == 42

The examples use top-level await, as in python -m asyncio; in an app they live inside async defs. Every example in this README runs in CI.

A handler serves the request its parameter is hinted with, or the one named with @handler(PlaceOrder). Each request has exactly one handler. Scanning is all or nothing: it reports every problem it finds — a missing hint, a duplicate handler, a module that fails to import — in one ScanError, and registers nothing.

Prefer explicit wiring? mediator.register(PlaceOrder, PlaceOrderHandler) does the same for one handler, and never needs a decorator.

Function handlers and dependency injection

A handler can be an async function. Its parameters after the request are dependencies, resolved by type hint on every call:

from dataclasses import dataclass

from mediary import Returns, handler, request


class Inventory:
    def __init__(self) -> None:
        self.counts = {"book": 3}


@request
@dataclass
class CheckStock(Returns[int]):
    item: str


@handler
async def check_stock(request: CheckStock, inventory: Inventory) -> int:
    return inventory.counts.get(request.item, 0)

Handler classes and dependencies come from the mediator's resolver, which calls cls() by default. Plug in any DI container by adapting it to one method, resolve(cls), which may be sync or async:

from shop.stock import CheckStock, Inventory


class ContainerResolver:
    def __init__(self) -> None:
        self.singletons = {Inventory: Inventory()}

    def resolve(self, cls):
        return self.singletons.get(cls) or cls()


mediator = Mediator(resolver=ContainerResolver())
mediator.scan("shop")
assert await mediator.send(CheckStock("book")) == 3

A class handler is resolved for every send, unless it is decorated @handler(lifetime="singleton").

Behaviors

Behaviors wrap handlers like middleware: each gets the message and next, and can act before and after it, change the result, or skip the handler.

from mediary import Next, behavior

calls = []


@behavior(order=-10)  # lower orders run further out
async def trace(request: object, next: Next[object]) -> object:
    calls.append(f"-> {type(request).__name__}")
    result = await next()
    calls.append(f"<- {result}")
    return result


@behavior
async def double_orders(request: PlaceOrder, next: Next[int]) -> int:
    return 2 * await next()


mediator = Mediator()
mediator.scan("shop")
mediator.use(trace)  # scan finds decorated behaviors too; `use` adds them by hand
mediator.use(double_orders)

assert await mediator.send(PlaceOrder("book", 1)) == 84
assert await mediator.send(CheckStock("book")) == 3  # double_orders only wraps PlaceOrder
assert calls == ["-> PlaceOrder", "<- 84", "-> CheckStock", "<- 3"]

The hint on the request parameter picks what a behavior wraps: object for everything, a class for it and its subclasses, a Protocol for every request with those members, or a union. kinds={"request"} narrows it to kinds of message, and lower orders run further out (ties are broken by name).

Three ready-made behaviors cover production basics. They are never scanned; add them configured:

from mediary.behaviors import LoggingBehavior, RetryBehavior, TimeoutBehavior

mediator.use(LoggingBehavior(), order=-100)  # start, completion, failure, slowness
mediator.use(TimeoutBehavior(seconds=5), order=-50)  # HandlerTimeout when it's too slow
mediator.use(RetryBehavior(max_retries=3), kinds={"request"})  # backoff with jitter

RetryBehavior retries only transient errors — those whose class is marked @retryable, like every TransientError — so a bug never runs twice:

from mediary import TransientError, retryable


class GatewayUnavailable(TransientError):  # retried
    pass


@retryable
class StorageError(Exception):  # retried, and so are its subclasses
    pass


class CardDeclined(Exception):  # fails at once
    pass

Errors you can't decorate, such as ConnectionError, can be listed: RetryBehavior(retry_on=(ConnectionError,)).

Notifications

A notification goes to every one of its handlers — zero or more — in order of their names:

from dataclasses import dataclass

from mediary import Concurrent, notification


@notification
@dataclass
class OrderPlaced:
    order_id: int


emails = []


async def email_customer(event: OrderPlaced) -> None:
    emails.append(f"order {event.order_id} confirmed")


async def update_stats(event: OrderPlaced) -> None:
    pass


mediator = Mediator()
mediator.register(OrderPlaced, email_customer)
mediator.register(OrderPlaced, update_stats)

await mediator.publish(OrderPlaced(42))  # one handler after the other
await mediator.publish(OrderPlaced(42), strategy=Concurrent())  # all at once
assert emails == ["order 42 confirmed"] * 2

Concurrent runs every handler even when some fail, then raises their errors together in an ExceptionGroup. Pass Mediator(publish_strategy=...) to change the default, or write your own strategy.

CQRS

mediary.cqrs speaks the language of CQRS: commands change state, queries read it, and events announce what happened.

from dataclasses import dataclass

from mediary.cqrs import Command, Query, QuerySender, command, query

names = {}


@command
@dataclass
class RenameUser(Command[None]):
    user_id: int
    name: str


@query
@dataclass
class GetUserName(Query[str]):
    user_id: int


async def rename_user(command: RenameUser) -> None:
    names[command.user_id] = command.name


async def get_user_name(query: GetUserName) -> str:
    return names[query.user_id]


mediator = Mediator()
mediator.register(RenameUser, rename_user)
mediator.register(GetUserName, get_user_name)


async def profile_page(queries: QuerySender, user_id: int) -> str:
    # A QuerySender can't send commands: type checkers reject `queries.send(RenameUser(...))`.
    return f"<h1>{await queries.send(GetUserName(user_id))}</h1>"


await mediator.send(RenameUser(1, "Ada"))
assert await profile_page(mediator, 1) == "<h1>Ada</h1>"

Each command and query has exactly one handler, and a query handler annotated to return None is rejected. Behaviors can target kinds={"command"}, {"query"} or {"event"}.

The pack is built only on the public mediary.kinds API, which you can use to define kinds of your own, with rules their handlers must follow:

from mediary import Returns
from mediary.kinds import HandlerInfo, define_kind


def returns_something(info: HandlerInfo) -> str | None:
    if info.returns is type(None):
        return "a report must return its rows"
    return None


report = define_kind("report", dispatch="send", rules=[returns_something])


@report
class SalesByMonth(Returns[list[int]]):
    pass

Testing

mediary.testing.RecordingMediator is a Mediator that records what it sends and publishes, and can answer requests with stubs. With mediary installed, pytest provides a fresh one as the mediator fixture:

from mediary.testing import RecordingMediator


async def place_and_announce(mediator: Mediator, item: str) -> None:
    order_id = await mediator.send(PlaceOrder(item, 1))
    await mediator.publish(OrderPlaced(order_id))


async def test_placing_an_order_announces_it(mediator: RecordingMediator) -> None:
    mediator.stub(PlaceOrder, 7)

    await place_and_announce(mediator, "book")

    assert mediator.sent_of(PlaceOrder) == [PlaceOrder("book", 1)]
    assert mediator.published_of(OrderPlaced) == [OrderPlaced(7)]

Stubs stand in for handlers — mediator.stub(PlaceOrder, raises=CardDeclined()) fails instead — and behaviors still wrap them. Every mediator is isolated, so tests never share registrations.

Why not register by hand?

Most mediator libraries have you register each request with its handler, and each pipeline step, in one central place that every feature has to edit. With mediary:

register and use are still there when you want explicit wiring, as in libraries and tests.

Development

Requires uv. See CONTRIBUTING.md for the conventions.

uv sync                        # create .venv with dev tools
uv run pre-commit install      # lint, format and typecheck on commit
uv run pytest                  # tests + coverage gate (95%)
uv run pyright                 # strict type checking

License

MIT