The mediator

How send, stream, and publish resolve handlers, apply behaviors, and propagate errors.

Mediator connects message declarations to the handler instances in a ServiceProvider. It supports three dispatch forms:

Method	Input	Handler count	Result	Pipeline behaviors
`await send(request)`	`Request[T]`	One	`T`	Applied
`stream(request)`	`StreamRequest[T]`	One	`AsyncIterator[T]`	Not applied
`await publish(notification)`	`Notification`	Zero or more	`None`	Not applied

The top-level pymediate.Mediator is asynchronous. The pymediate.sync variant provides the corresponding blocking methods.

Build a mediator

from pymediate import Mediator, Services

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

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

Services is read-only once constructed, so a mediator's resolution can never change underneath it. The mediator constructor also accepts another implementation of the ServiceProvider protocol, such as the optional dependency-injector adapter.

How `send()` resolves a request

Each send() call follows the same sequence:

The exact runtime class of the request identifies its request-handler class. This mapping was recorded when Python defined RequestHandler[RequestType].
The service provider resolves an instance of that handler class.
The provider resolves registered PipelineBehavior instances. The mediator keeps the ones whose should_apply() method accepts the request.
If behaviors apply, the mediator composes them around the handler. The first registered behavior is the outermost. If none apply, it calls the handler directly without constructing a behavior chain.
The mediator awaits the handler path and returns its response.

The return type comes from the request declaration:

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

PyMediate validates the handler's parameter and return annotations when the handler class is defined. send() does not inspect the type of the value returned at runtime.

Pipeline behaviors apply only to send(). They can run before and after the handler, or return a response without calling the handler. The pipeline-behavior guide covers selection, ordering, and short-circuiting.

How `stream()` resolves a stream

stream() uses the same exact-type lookup and service resolution for a StreamRequestHandler. Resolution happens when stream() is called, while the generator body is lazy:

chunks = mediator.stream(ExportOrders(customer_id=7))  # resolves the handler

async for chunk in chunks:  # runs the handler body as chunks are requested
    print(chunk)

Do not await mediator.stream() itself. The asynchronous method returns an AsyncIterator[T] for use with async for; the synchronous method returns an Iterator[T] for use with for.

Pipeline behaviors do not wrap streams. The streaming guide explains the generator contract and definition-time validation.

How `publish()` resolves subscribers

publish() looks up subscribers by the notification's exact runtime class. Publishing a notification with no subscribers completes without error.

For a notification with subscribers, the mediator resolves every handler instance before running any of them. The asynchronous mediator then runs the handlers concurrently. The synchronous mediator runs them sequentially in registration order. Pipeline behaviors do not wrap either form.

The asynchronous mediator lets every subscriber finish after ordinary Exception failures and then reports those failures as a group. KeyboardInterrupt and SystemExit propagate instead and can cancel unfinished subscribers. The synchronous mediator continues only after ordinary Exception failures; a direct BaseException stops delivery and propagates immediately. See notifications for the concurrency and error details.

Handler and behavior lifetimes

The service provider controls instance lifetimes; the mediator does not cache resolved services.

With Services, the provider returns the instances supplied during registration. Reusing that provider therefore reuses those handler and behavior instances.
With DependencyInjectorServiceProvider, a Factory can create a new instance on each resolution, while a Singleton can return one shared instance.

A mediator can normally be built during application startup and reused. Whether its handlers are safe to share or call concurrently depends on their own state and dependencies. The handlers guide covers that distinction.

Errors and timing

Resolution and user-code errors occur at different points:

Situation	Result
No handler class is declared for a request or stream request	`HandlerNotFoundError` at `send()` or `stream()`
A handler class is declared but its instance is absent from the provider	`ServiceNotFoundError` at `send()` or `stream()`
A subscriber class is declared but its instance is absent	`ServiceNotFoundError` before any subscriber runs
A request handler or behavior raises	The exception propagates unchanged from `send()`
A stream handler raises in its generator body	The exception propagates during iteration
One or more async notification handlers raise an ordinary `Exception`	Failures are grouped after every subscriber finishes
An async notification handler raises `KeyboardInterrupt` or `SystemExit`	The failure propagates and can cancel unfinished subscribers
One or more sync notification handlers raise an ordinary `Exception`	Failures are grouped after every subscriber has been called
A sync notification handler raises a direct `BaseException`	Delivery stops and the failure propagates immediately

The error-handling guide distinguishes these library errors from application and transport errors.

Use the synchronous mediator

The blocking API has the same three methods without await or asynchronous iteration:

from pymediate.sync import Mediator, Services

mediator = Mediator(Services(PlaceOrderHandler()))

receipt = mediator.send(PlaceOrder(customer_id=7, item="tea", quantity=2))
for chunk in mediator.stream(ExportOrders(customer_id=7)):
    print(chunk)
mediator.publish(OrderPlaced(order_id=receipt.order_id, item="tea"))

Use synchronous handler and behavior classes with the synchronous mediator. The two mediator variants do not adapt handler call styles at runtime. See async and sync for the corresponding imports and coexistence rules.

Continue

Add shared request processing with pipeline behaviors.
Configure service lifetimes with dependency injection.
Look up the exact Mediator signatures.

On this page