Pipeline behaviors

A PipelineBehavior runs around request handling. It can inspect a request, run code before and after the next step, modify the response, or return without calling the handler.

Behaviors wrap Mediator.send() only. They do not wrap stream() or publish().

Define a typed behavior

The behavior's type argument selects its request type. Its next parameter represents the remaining behaviors and, finally, the request handler:

from typing import override

from pymediate import Next, PipelineBehavior


class ValidatePlaceOrder(PipelineBehavior[PlaceOrder]):
    @override
    async def __call__(
        self,
        request: PlaceOrder,
        next: Next[OrderReceipt],
    ) -> OrderReceipt:
        if request.quantity < 1:
            raise ValueError("quantity must be at least 1")

        return await next()

Read PipelineBehavior[PlaceOrder] as “this behavior applies to PlaceOrder requests.” Next[OrderReceipt] records the response returned by the rest of this particular pipeline.

@override lets a type checker confirm that __call__ matches the asynchronous behavior API. PyMediate does not apply the request-handler class-definition checks to behavior methods, so static checking is useful here.

Follow the call order

Behaviors form nested calls. The first entry in the mediator's behaviors sequence is the outermost:

Request
  -> RequestLogging
    -> ValidatePlaceOrder
      -> PlaceOrderHandler
      <- OrderReceipt
    <- OrderReceipt
  <- OrderReceipt

Calling await next() enters the next layer. Code after it runs as the response returns through the layers. A try, except, or finally block can observe failures from every inner behavior and the handler.

If a behavior does not call next(), the remaining layers do not run. This supports deliberate short-circuits such as a cache hit or an authorization rejection. A behavior that returns a value directly should still return the response type declared by the request.

Calling next() more than once repeats every inner layer, including the handler. A retry behavior therefore needs the same idempotency and transaction-ordering decisions as any other retry mechanism.

Apply a behavior to every request

Use Request[Any] for a behavior that accepts all request types. Keep all three annotations broad because different requests can return different response types:

from typing import Any, override

from pymediate import Next, PipelineBehavior, Request


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

The PipelineBehavior type parameter is bounded to Request[Any]. Use Request[Any] or a concrete Request subclass, rather than an unrelated mixin type, so the declaration satisfies the public typing contract.

Select request types

PipelineBehavior[PlaceOrder] applies to PlaceOrder and its subclasses by default. Set the public apply_to_subclasses class attribute to False for exact-type matching:

class ExactPlaceOrderValidation(PipelineBehavior[PlaceOrder]):
    apply_to_subclasses = False

    @override
    async def __call__(
        self,
        request: PlaceOrder,
        next: Next[OrderReceipt],
    ) -> OrderReceipt:
        if request.quantity < 1:
            raise ValueError("quantity must be at least 1")
        return await next()

For selection based on request values, override should_apply(). An override replaces the default type test, so include the type condition explicitly:

class AuditBulkOrders(PipelineBehavior[PlaceOrder]):
    @classmethod
    @override
    def should_apply(cls, request: Request[Any]) -> bool:
        return isinstance(request, PlaceOrder) and request.quantity >= 10

    @override
    async def __call__(
        self,
        request: PlaceOrder,
        next: Next[OrderReceipt],
    ) -> OrderReceipt:
        print(f"auditing order of {request.quantity} items")
        return await next()

The mediator calls should_apply() for each registered behavior on every send().

Register and order behaviors

Register behaviors through the same provider as handlers, then declare the pipeline explicitly with the mediator's behaviors argument - an ordered sequence of behavior classes, first entry outermost:

from pymediate import Mediator, Services

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

behaviors is the entire pipeline declaration - it does not have to match registration order, and a behavior registered with the provider but left out of behaviors is simply not part of this mediator's pipeline. Omitting behaviors (or passing None) means no behaviors run, even if some are registered. The mediator validates the sequence when it is constructed: every entry must be a PipelineBehavior subclass of the mediator's variant, registered with the provider, and listed at most once - an invalid entry raises InvalidPipelineBehaviorsError immediately, before any send(). When no listed behavior accepts a request, the mediator calls the handler directly and does not construct a behavior chain.

Ordering changes semantics. For example, a transaction outside a retry covers every attempt in one transaction, while a transaction inside a retry can create one transaction per attempt. Define the intended order in behaviors and cover it with a mediator-level test.

The provider controls behavior lifetimes; behaviors only controls which classes run and in what order. Services reuses the registered instances. DependencyInjectorServiceProvider can resolve behaviors through Factory, Singleton, or other supported container lifetimes.

Use the synchronous behavior API

Import PipelineBehavior and Next from pymediate.sync, implement def __call__, and call next() without await:

from typing import override

from pymediate.sync import Next, PipelineBehavior


class ValidatePlaceOrder(PipelineBehavior[PlaceOrder]):
    @override
    def __call__(
        self,
        request: PlaceOrder,
        next: Next[OrderReceipt],
    ) -> OrderReceipt:
        if request.quantity < 1:
            raise ValueError("quantity must be at least 1")
        return next()

Asynchronous and synchronous behaviors are separate service types. Register the form that matches the mediator and handler being used.

Test a behavior directly

A behavior is a callable, so a focused test can supply a small next function:

async def finish() -> OrderReceipt:
    return OrderReceipt(order_id=42, summary="2 × tea")


behavior = ValidatePlaceOrder()
receipt = await behavior(
    PlaceOrder(customer_id=7, item="tea", quantity=2),
    finish,
)

assert receipt.order_id == 42

Use a mediator-level test when behavior selection, pipeline order, or provider lifetimes are part of the behavior under test.

Continue

• Review behavior resolution in the mediator.
• Configure container-managed lifetimes with dependency injection.
• Look up PipelineBehavior, Next, and apply_to_subclasses.