pymediate
API Reference

Pipeline

API reference for PipelineBehavior and Pipeline, sync and async.

Pipeline behaviors wrap request processing with middleware-style logic; a Pipeline chains behaviors around a handler. Both exist in sync (pymediate / pymediate.pipeline) and async (pymediate.aio) variants.

PipelineBehavior

from collections.abc import Callable
from typing import Any
from pymediate import PipelineBehavior, Request

class PipelineBehavior[RequestT: Request[Any]](ABC):
    @classmethod
    def should_apply(cls, request: Request[Any]) -> bool:
        """Determine if this behavior should apply to the given request."""

    @abstractmethod
    def __call__(self, request: RequestT, next: Callable[[], Any]) -> Any:
        """Execute the behavior's logic and call next to continue the pipeline."""
from collections.abc import Awaitable, Callable
from typing import Any
from pymediate.aio import PipelineBehavior

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

    @abstractmethod
    async def __call__(self, request: RequestT, next: Callable[[], Awaitable[Any]]) -> Any:
        """Execute the behavior's async logic and await next to continue the pipeline."""

Type parameters

ParameterDescription
RequestTWhich requests this behavior wraps: Request for all, a specific request type, or a mixin

__call__(request, next)

The behavior's body. next is a callable representing the rest of the pipeline — remaining behaviors and, ultimately, the handler. Call it (or await it in the async variant) to continue; skip it to short-circuit.

Argsrequest — the request being processed; next — continues the pipeline and returns the response
ReturnsThe response (usually the result of next(), possibly transformed)

should_apply(request) classmethod

Decides whether the behavior wraps a given request. The default implementation is an isinstance check against RequestT. Override it for matching logic beyond types:

class BusinessHoursBehavior(PipelineBehavior[Request]):
    @classmethod
    def should_apply(cls, request: Request) -> bool:
        return 9 <= datetime.now().hour < 17

Pipeline

from pymediate.pipeline import Pipeline

class Pipeline[RequestT, ResponseT]:
    def __init__(self, behaviors: Sequence[Any], handler: Handler[RequestT]) -> None:
        """Initialize a pipeline with behaviors and a handler."""

    def __call__(self, request: RequestT) -> ResponseT:
        """Process a request through the pipeline."""
from pymediate.aio.pipeline import Pipeline

class Pipeline[RequestT, ResponseT]:
    def __init__(self, behaviors: Sequence[Any], handler: Handler[RequestT]) -> None: ...

    async def __call__(self, request: RequestT) -> ResponseT: ...

Combines behaviors and a handler into one callable. Behaviors execute in the order given — first is outermost:

from pymediate.pipeline import Pipeline

pipeline = Pipeline(
    behaviors=[
        LoggingBehavior(logger),    # outermost
        ValidationBehavior(),
        TimingBehavior(metrics),    # innermost
    ],
    handler=GetUserHandler(database),
)

response = pipeline(GetUserRequest(user_id=123))

You rarely construct a Pipeline yourself — the mediator builds one per request from registered behaviors automatically. Manual construction is useful for testing behavior combinations and one-off workflows. See pipeline behaviors guide.

See also

Mediator

API reference for the sync and async Mediator classes.

Service provider

API reference for the ServiceProvider protocol, the Services collection, and the dependency-injector integration.

On this page

PipelineBehaviorType parameters__call__(request, next)should_apply(request) classmethodPipelineSee also