Request handlers

A request handler implements one operation for one request type. It receives the request, uses any supplied dependencies, and returns the response declared by that request.

Define a handler

Subclass RequestHandler[RequestType] and implement __call__:

from pymediate import RequestHandler

class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        return OrderReceipt(
            order_id=42,
            summary=f"{request.quantity} × {request.item}",
        )

PlaceOrderHandler(RequestHandler[PlaceOrder]) means that the class handles PlaceOrder requests. The request's Request[OrderReceipt] base supplies the expected return annotation.

Match the declared request exactly

The method parameter must use the same concrete request type as the handler's generic argument:

class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        ...

A base class, protocol, union, or Any does not satisfy PyMediate's runtime contract for this parameter. Dispatch uses the exact class of the request instance, so a broader handler annotation would describe values that this registration cannot receive.

Understand definition-time validation

When Python defines a request-handler subclass, PyMediate checks that:

• its generic argument is a Request subclass;
• __call__ has one annotated request parameter;
• the parameter annotation is the declared request type;
• the return annotation matches the request's response type;
• the method is asynchronous for pymediate.RequestHandler or synchronous for pymediate.sync.RequestHandler;
• another request-handler type has not already been registered for that request type.

These checks validate annotations and registration. They do not execute the handler or inspect the value it later returns.

See type safety for the division between static checking, class-definition validation, and dispatch-time errors.

Supply operational dependencies

Constructor parameters make a handler's dependencies visible and replaceable:

from typing import Protocol

class OrderStore(Protocol):
    async def add(self, request: PlaceOrder) -> int: ...

class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    def __init__(self, orders: OrderStore) -> None:
        self._orders = orders

    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        order_id = await self._orders.add(request)
        return OrderReceipt(
            order_id=order_id,
            summary=f"{request.quantity} × {request.item}",
        )

Constructing infrastructure inside the handler is valid Python, but it couples the operation to that implementation and makes substitution in tests or other deployments harder. Supply a dependency when its implementation or lifetime needs to vary.

The built-in Services collection stores instances. The optional Dependency Injector integration can resolve handlers with container-managed lifetimes.

Keep state ownership explicit

A handler without mutable fields can be reused, but that alone does not make the operation thread-safe or task-safe. Its supplied dependencies may own connections, caches, transactions, or other mutable state with their own concurrency rules.

If a handler stores in-memory state, document the intended lifetime and concurrency:

class CountingHandler(RequestHandler[CountOrder]):
    def __init__(self) -> None:
        self._count = 0

    async def __call__(self, request: CountOrder) -> CountResult:
        self._count += 1
        return CountResult(value=self._count)

That counter belongs to one handler instance. It is lost when the process stops and needs synchronization if calls can overlap. Persistent business state normally belongs in an explicitly supplied store.

Compose operations deliberately

One handler can call another operation through an ordinary collaborator:

class RefundOrderHandler(RequestHandler[RefundOrder]):
    def __init__(self, payments: PaymentRefunds) -> None:
        self._payments = payments

This direct dependency follows an ordinary object reference and is often sufficient.

A handler can instead receive a mediator when it must dispatch another application request without depending on that request's handler:

class RefundOrderHandler(RequestHandler[RefundOrder]):
    def __init__(self, mediator: Mediator) -> None:
        self._mediator = mediator

    async def __call__(self, request: RefundOrder) -> RefundResult:
        payment = await self._mediator.send(
            RefundPayment(payment_id=request.payment_id),
        )
        return RefundResult(refund_id=payment.refund_id)

This replaces a direct operation dependency with request and mediator dependencies. Use it when that indirection is useful, and avoid request cycles such as A → B → A.

Test handlers directly

Handlers are callable objects, so most operation tests do not need a mediator:

async def test_place_order() -> None:
    orders = InMemoryOrderStore()
    handler = PlaceOrderHandler(orders)

    receipt = await handler(
        PlaceOrder(customer_id=7, item="tea", quantity=2),
    )

    assert receipt.order_id == 1

Use mediator-level tests for registration, service resolution, pipeline ordering, and complete dispatch. The testing guide covers both levels and the process-wide handler registry.

Use the synchronous handler when needed

Import RequestHandler from pymediate.sync and implement def __call__ for blocking code. Do not define asynchronous and synchronous handlers for the same request type in one process; the request-handler registry allows one handler type per request type.

See async and sync for selection and coexistence rules.

Continue

• Configure dispatch with the mediator.
• Add shared request processing with pipeline behaviors.
• Look up the exact RequestHandler API contract.