Typed in-process request dispatch for Python 3.12+

---

PyMediate routes typed requests to handlers. A request declares its response type, so Mediator.send() preserves that type for static type checkers and editors.

Installation

pip install pymediate

The core package has no required dependencies. Install the optional Dependency Injector integration with pip install 'pymediate[di]'.

First request

import asyncio
from dataclasses import dataclass

from pymediate import Mediator, Request, RequestHandler, Services


@dataclass(frozen=True)
class OrderReceipt:
    order_id: int
    summary: str


@dataclass(frozen=True)
class PlaceOrder(Request[OrderReceipt]):
    customer_id: int
    item: str
    quantity: int


class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        return OrderReceipt(
            order_id=42,
            summary=f"{request.quantity} × {request.item}",
        )


async def main() -> None:
    mediator = Mediator(Services(PlaceOrderHandler()))
    receipt = await mediator.send(
        PlaceOrder(customer_id=7, item="tea", quantity=2),
    )
    print(receipt.order_id, receipt.summary)


asyncio.run(main())

The program prints:

42 2 × tea

Read the declarations as follows:

• OrderReceipt is the response.
• PlaceOrder(Request[OrderReceipt]) is a request for an OrderReceipt.
• PlaceOrderHandler(RequestHandler[PlaceOrder]) handles PlaceOrder requests.

The introduction contains an interactive guide to these relationships. The quick start explains the complete dispatch flow.

API at a glance

Need	API	Result
Send a request to one handler	Mediator.send()	One response, inferred from Request[T]
Yield results over time	Mediator.stream()	Typed chunks from StreamRequest[T]
Notify zero or more subscribers	Mediator.publish()	No response
Wrap request handling	PipelineBehavior	Shared processing around send()
Supply handlers and behaviors	Services or another ServiceProvider	Resolved instances

The top-level package is asynchronous. pymediate.sync provides corresponding blocking mediator and handler classes. Shared message types, services, and errors are the same objects in both namespaces.

Type checking and validation

Request[T] records the return type used by send() at static call sites. Separately, PyMediate checks a request handler's parameter annotation, return annotation, and asynchronous or synchronous form when Python defines the handler class.

Configuration can still fail during dispatch. For example, sending a request without a registered handler instance raises an error at that point. The type-safety guide describes which checks happen statically, at class definition, and during dispatch.

Scope and trade-offs

PyMediate provides in-process dispatch. It does not provide a task queue, choose persistence, or require CQRS or hexagonal architecture. Direct calls are often clearer when callers can depend on their collaborators without repeated wiring, and a small hand-written dispatcher can be enough.

The article Using a mediator to reduce change coupling develops the case for a mediator and covers the added indirection, registration, and runtime cost. The comparison documents the current feature set, dated dispatch benchmarks, and a runnable benchmark script.

Documentation

• Introduction
• Quick start
• Core concepts
• Guides
• API reference
• Runnable examples

Development

git clone https://github.com/sina-al/pymediate.git
cd pymediate
uv sync --all-extras --group test

uv run poe test
uv run poe check:all

Run uv run poe to list the repository tasks. See CONTRIBUTING.md for the contribution process.

Versioning

PyMediate follows ZeroVer: the major version remains 0. A minor release (0.X.0) can contain a breaking API change or a backward-compatible feature. A patch release (0.X.Y) contains changes that do not alter the public API.

License

PyMediate is available under the MIT License.