Mediator

Mediator resolves handler instances through a ServiceProvider and supports three dispatch forms: one response, a stream of chunks, or notification publication.

from collections.abc import AsyncIterator

from collections.abc import Sequence

from pymediate import Notification, PipelineBehavior, Request, ServiceProvider, StreamRequest

class Mediator:
    def __init__(
        self,
        services: ServiceProvider,
        *,
        behaviors: Sequence[type[PipelineBehavior]] | None = None,
    ) -> None: ...

    async def send[ResponseT](self, request: Request[ResponseT]) -> ResponseT: ...

    def stream[ChunkT](
        self,
        request: StreamRequest[ChunkT],
    ) -> AsyncIterator[ChunkT]: ...

    async def publish(self, notification: Notification) -> None: ...

Constructor

Mediator(services, behaviors=None) accepts any object implementing the ServiceProvider protocol, plus the ordered PipelineBehavior classes that make up the pipeline. services and behaviors are both keyword names:

from pymediate import Mediator, Services

services = Services(RequestLogging(), PlaceOrderHandler())
mediator = Mediator(services=services, behaviors=[RequestLogging])

behaviors declares the pipeline explicitly: an ordered sequence of behavior classes, first entry outermost. A behavior registered with the provider but absent from behaviors is not part of the pipeline. Omitting behaviors (or passing None, the default) means no behaviors run.

The mediator validates behaviors once, at construction: every entry must be a PipelineBehavior subclass of the mediator's variant, registered with the provider (checked via in), and listed at most once. A violation raises InvalidPipelineBehaviorsError immediately, naming the offending entry, before any send().

The mediator keeps the provider for later dispatches. Handler and behavior lifetimes are therefore determined by that provider.

send()

send() uses the request's exact class to find its registered RequestHandler class, resolves a handler instance, and calls it. It returns the response type declared by Request[ResponseT].

Before the handler runs, the mediator walks the constructor's behaviors sequence in order and keeps the entries whose should_apply() method returns True, resolving each through the provider. The first listed applicable behavior is the outermost. If none apply, the mediator calls the handler without constructing a behavior chain.

Exceptions raised by a behavior or handler propagate unchanged.

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

stream()

stream() uses the stream request's exact class to resolve its StreamRequestHandler and returns the resulting iterator. The handler instance is resolved at the stream() call. The generator body is lazy and runs as the caller consumes chunks.

Errors from inside the generator are raised during iteration. Pipeline behaviors do not wrap stream().

async for row in mediator.stream(ExportOrders(customer_id=7)):
    print(row)

Do not await the asynchronous stream() call itself; use async for on its result.

publish()

publish() resolves every NotificationHandler registered for the notification's exact class. Publishing with no registered handlers returns None without error. Pipeline behaviors do not wrap publication.

The mediator resolves every handler instance before invoking any of them. A missing instance thus raises ServiceNotFoundError before partial delivery can occur.

The asynchronous mediator runs the resolved handlers concurrently. Ordinary Exception failures are collected after all handlers finish and raised as an ExceptionGroup. Other collected BaseException values produce a BaseExceptionGroup. Python treats KeyboardInterrupt and SystemExit specially: they propagate instead of being grouped and can cancel unfinished sibling handlers.

The synchronous mediator runs handlers sequentially in registration order, continues after ordinary Exception failures, and raises those failures together as an ExceptionGroup. A direct BaseException that is not an Exception stops synchronous delivery and propagates immediately.

await mediator.publish(OrderPlaced(order_id=42, item="tea"))

See also

• RequestHandler — the send() handler
• StreamRequestHandler — the stream() handler
• NotificationHandler — a publish() subscriber
• PipelineBehavior and Next — request middleware
• Errors — dispatch and publication failures