Python 3.12+ async and sync MIT

A typed mediator for Python

PyMediate routes in-process requests to handlers. Each request declares its response type, so static type checkers and editors infer what send() returns.

Read the introductionQuick start
callerPlaceOrdermediatorPlaceOrderHandlerhandlerOrderReceipt

One request, one handler, one response type

PlaceOrder declares that it returns an OrderReceipt. The handler accepts that request, and the mediator dispatches it. The Behavior and Notification tabs are focused excerpts built on the same shop domain.

See how to read the types
async_request.py
import asyncio
from dataclasses import dataclass
from typing import override

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]):
    @override
    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        return OrderReceipt(
            order_id=42,
            summary=f"{request.quantity} × {request.item}",
        )

async def main() -> None:
    services = Services(PlaceOrderHandler())

    mediator = Mediator(services)
    request = PlaceOrder(customer_id=7, item="tea", quantity=2)
    receipt = await mediator.send(request)

    print(receipt.summary)

asyncio.run(main())

What the package provides

Start with a request and its handler. Use streams for results over time, notifications for zero-or-more subscribers, behaviors for shared processing, and a service provider to resolve the registered instances.

Declared response types

A Request[T] declaration gives send() its return type. Type checkers and editors preserve that relationship at each call site.

Checked handler annotations

PyMediate checks request-handler parameter and return annotations when Python defines the handler class.

Asynchronous and synchronous APIs

The top-level package is asynchronous. pymediate.sync provides the corresponding blocking mediator and handler classes.

Requests, streams, and notifications

send() returns one response, stream() yields typed chunks, and publish() delivers a notification to zero or more subscribers.

Pipeline behaviors

Behaviors wrap send() with shared processing such as logging, validation, caching, or transaction management.

Built-in or custom service providers

Use the built-in Services collection or another ServiceProvider implementation. The core package has no required dependencies.

Read the core concepts

Using a mediator to reduce change coupling

Direct calls are often the clearest design. A mediator becomes relevant when repeated handler lookup and shared processing spread across callers. The article follows that progression, including the cases where direct calls or a small dictionary remain sufficient.

Read the article

Build the first request flow

The quick start defines PlaceOrder, runs its handler through a mediator, and prints the returned receipt.

Quick startAPI reference