PipelineBehavior and Next

PipelineBehavior wraps request processing before and after the next behavior or handler. Behaviors apply to Mediator.send() only; they do not wrap stream() or publish().

Signatures

from abc import ABC, abstractmethod
from collections.abc import Awaitable, Callable
from typing import Any

from pymediate import Request

type Next[ResponseT] = Callable[[], Awaitable[ResponseT]]

class PipelineBehavior[RequestT: Request[Any]](ABC):
    apply_to_subclasses: bool = True

    @classmethod
    def should_apply(cls, request: Request[Any]) -> bool: ...

    @abstractmethod
    async def __call__(
        self,
        request: RequestT,
        next: Next[Any],
    ) -> Any: ...

The RequestT bound is Request[Any]. Use PipelineBehavior[Request[Any]] for a universal behavior or a specific request class such as PipelineBehavior[PlaceOrder] for a selective one.

__call__(request, next)

next is a zero-argument continuation. Await it in the asynchronous API, or call it directly in the synchronous API, to run the rest of the chain. A behavior may return without calling next() to short-circuit dispatch. Whatever the behavior returns becomes the result seen by the preceding behavior or caller.

For a behavior scoped to one request type, use the concrete response type on Next and on the method return:

from pymediate import Next, PipelineBehavior


class TracePlaceOrder(PipelineBehavior[PlaceOrder]):
    async def __call__(
        self,
        request: PlaceOrder,
        next: Next[OrderReceipt],
    ) -> OrderReceipt:
        print("handling PlaceOrder")
        receipt = await next()
        print("handled PlaceOrder")
        return receipt

For a universal behavior that can see different response types, use Next[Any] and return Any. PyMediate does not validate a behavior's returned value against the request's declared response type.

Selection

should_apply(request) decides whether a registered behavior joins a request's pipeline. The default behavior is:

• PipelineBehavior[Request[Any]] matches every Request;
• a behavior parameterized with a request class matches that class and its subclasses; and
• setting apply_to_subclasses = False restricts the match to the exact request class.

Override should_apply() when selection depends on request data or another condition.

Registration and order

Register behavior instances in the same ServiceProvider as request handlers, then declare the pipeline explicitly with Mediator's behaviors argument - an ordered sequence of behavior classes, first entry outermost:

services = Services(LoggingBehavior(), TracePlaceOrder(), PlaceOrderHandler())

mediator = Mediator(services, behaviors=[LoggingBehavior, TracePlaceOrder])

A behavior registered with the provider but left out of behaviors is not part of the pipeline; omitting behaviors means no behaviors run. See Mediator's constructor for validation details.

Use the asynchronous behavior class with pymediate.Mediator and the synchronous class with pymediate.sync.Mediator.

Direct invocation

There is no public pipeline object. A behavior can be tested directly by supplying a continuation:

receipt = await TracePlaceOrder()(request, lambda: handler(request))

See also

• Pipeline behaviors — selection, ordering, and design
• Mediator.send() — discovers and composes behaviors
• Testing — direct handler and behavior tests