Core concepts

PyMediate starts with one flow: a caller sends a request, one handler processes it, and the mediator returns the request's declared response type. Notifications, streaming, and pipeline behaviors extend that model for other needs.

Requests and responses

A request is a value that describes an operation. Its type parameter declares the response:

from dataclasses import dataclass
from pymediate import Request

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

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

Read PlaceOrder(Request[OrderReceipt]) as “PlaceOrder is a request for an OrderReceipt.” The response is an ordinary Python type; it does not inherit from PyMediate.

Request handlers

A request handler declares which request it accepts and implements __call__:

from pymediate import RequestHandler

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

Read PlaceOrderHandler(RequestHandler[PlaceOrder]) as “PlaceOrderHandler handles PlaceOrder requests.” Each request type has one registered request-handler type.

PyMediate validates the handler's annotated parameter and return types when Python defines the class. The handler still owns the operation's implementation, dependencies, and errors.

The mediator

The mediator is the dispatch entry point. It receives a request and returns the handler's response:

from pymediate import Mediator, Services

services = Services(PlaceOrderHandler())
mediator = Mediator(services)

receipt = await mediator.send(
    PlaceOrder(customer_id=7, item="tea", quantity=2),
)
# receipt is inferred as OrderReceipt

The caller imports the request type and receives the response type. It does not locate or construct the handler.

Services and service providers

The mediator needs handler instances. The built-in Services collection takes them at construction and is itself the read-only ServiceProvider the mediator resolves through:

from pymediate import Mediator, Services

services = Services(PlaceOrderHandler())
mediator = Mediator(services)

ServiceProvider is a protocol. Applications can use another implementation, including the optional dependency-injector integration, as long as it provides the required methods.

Three dispatch forms

The mediator supports three result shapes:

Streaming

A stream request declares the type of each yielded chunk. Its handler is an asynchronous generator:

from collections.abc import AsyncIterator
from dataclasses import dataclass
from pymediate import StreamRequest, StreamRequestHandler

@dataclass(frozen=True)
class ExportOrders(StreamRequest[bytes]):
    customer_id: int

class ExportOrdersHandler(StreamRequestHandler[ExportOrders]):
    async def __call__(self, request: ExportOrders) -> AsyncIterator[bytes]:
        yield b"order_id,total_pence\n"
        yield b"42,2500\n"

mediator.stream() returns an AsyncIterator[bytes]. The caller decides when to consume each chunk. See streaming for resolution and iteration behavior.

Notifications

A notification records that something happened. It has no response type, and any number of notification handlers may subscribe:

from dataclasses import dataclass
from pymediate import Notification, NotificationHandler

@dataclass(frozen=True)
class OrderPlaced(Notification):
    order_id: int
    item: str

class RecordOrderMetric(NotificationHandler[OrderPlaced]):
    async def __call__(self, notification: OrderPlaced) -> None:
        print(f"recorded order {notification.order_id}")

# After supplying RecordOrderMetric to the mediator's service provider:
await mediator.publish(OrderPlaced(order_id=42, item="tea"))

Publishing with no subscribers is valid. The notifications guide covers concurrency, ordering, and error aggregation.

Pipeline behaviors

Pipeline behaviors wrap request handling. They can run logic before and after the next behavior or handler:

from typing import Any
from pymediate import Next, PipelineBehavior, Request

class LoggingBehavior(PipelineBehavior[Request[Any]]):
    async def __call__(self, request: Request[Any], next: Next[Any]) -> Any:
        print(f"handling {type(request).__name__}")
        response = await next()
        print(f"handled {type(response).__name__}")
        return response

Behaviors can apply to every request or to selected request types. They wrap send(); they do not wrap stream() or publish(). See pipeline behaviors for selection and ordering.

Asynchronous and synchronous APIs

The top-level pymediate package is asynchronous. Request and notification handlers use async def, send() and publish() are awaited, and stream() returns an asynchronous iterator.

pymediate.sync provides corresponding blocking mediator and handler classes. Shared message types, services, and errors are the same objects in both namespaces.

Continue

• Design messages with requests and responses.
• Add dependencies and tests with the handlers guide.
• Check exact signatures in the API reference.