Handlers

Writing handlers — single responsibility, dependency injection, async, statelessness, and composition through the mediator.

Handlers are PyMediate's execution units. Each one holds the business logic for exactly one request type, independent of every other handler and of the infrastructure that invokes it.

Anatomy of a handler

A handler subclasses Handler[RequestType] and implements __call__:

from dataclasses import dataclass
from pymediate import Handler, Request

@dataclass
class CreateUserResponse:
    user_id: int
    username: str

@dataclass
class CreateUserRequest(Request[CreateUserResponse]):
    username: str
    email: str

class CreateUserHandler(Handler[CreateUserRequest]):
    def __init__(self, database):
        self.database = database

    def __call__(self, request: CreateUserRequest) -> CreateUserResponse:
        user_id = self.database.insert_user(username=request.username, email=request.email)
        return CreateUserResponse(user_id=user_id, username=request.username)

Four properties fall out of this shape:

Single responsibility — one handler, one request type.
Type safety — the Handler[RequestType] parameter links handler to request in the type system, and the return type must match the request's declared response.
Framework independence — nothing here knows about HTTP, CLIs, or queues.
Testability — it's a plain callable; construct it and call it.

Handlers never know about each other. They know their request, their response, and their own dependencies — nothing else. When one operation needs another, go through the mediator rather than holding a reference to the other handler:

# Tight coupling — don't do this
class CreateOrderHandler(Handler[CreateOrderRequest]):
    def __init__(self, create_user_handler, send_email_handler):
        self.create_user_handler = create_user_handler
        self.send_email_handler = send_email_handler

# Loose coupling through the mediator
class CreateOrderHandler(Handler[CreateOrderRequest]):
    def __init__(self, mediator, database):
        self.mediator = mediator
        self.database = database

    def __call__(self, request: CreateOrderRequest) -> CreateOrderResponse:
        user = self.mediator.send(CreateUserRequest(...))
        self.mediator.send(SendEmailRequest(...))
        order_id = self.database.create_order(user_id=user.user_id)
        return CreateOrderResponse(order_id=order_id)

Independence is what buys you deployment flexibility: the same handler can serve a Flask route today, an AWS Lambda tomorrow, and a Kafka consumer next month — only the thin adapter changes. See requests and responses for adapter examples.

Dependencies

Inject dependencies through the constructor; never construct them inside the handler:

class UserServiceHandler(Handler[GetUserRequest]):
    def __init__(self, database, cache, logger):
        self.database = database
        self.cache = cache
        self.logger = logger

    def __call__(self, request: GetUserRequest) -> GetUserResponse:
        cached = self.cache.get(f"user:{request.user_id}")
        if cached:
            return GetUserResponse(**cached)

        user = self.database.get_user(request.user_id)
        self.cache.set(f"user:{request.user_id}", user.to_dict())
        return GetUserResponse(user_id=user.id, username=user.username, email=user.email)

For wiring dependencies through a container, see dependency injection.

Async handlers

For I/O-bound work, use the async mirror in pymediate.aio. The only changes: the import and async def __call__:

from dataclasses import dataclass
from pymediate import Request
from pymediate.aio import Handler, Mediator

@dataclass
class FetchDataResponse:
    data: dict

@dataclass
class FetchDataRequest(Request[FetchDataResponse]):
    url: str

class AsyncFetchHandler(Handler[FetchDataRequest]):
    def __init__(self, http_client):
        self.http_client = http_client

    async def __call__(self, request: FetchDataRequest) -> FetchDataResponse:
        data = await self.http_client.get(request.url)
        return FetchDataResponse(data=data)

Sync vs async imports

Sync: from pymediate import Handler, Mediator. Async: from pymediate.aio import Handler, Mediator. Request always comes from the main package — the same request types work with both.

Async handlers can fan out concurrent I/O with asyncio.gather:

import asyncio
from pymediate.aio import Handler

class ProcessOrderHandler(Handler[ProcessOrderRequest]):
    def __init__(self, payment_service, inventory_service):
        self.payment_service = payment_service
        self.inventory_service = inventory_service

    async def __call__(self, request: ProcessOrderRequest) -> ProcessOrderResponse:
        payment_result, inventory_result = await asyncio.gather(
            self.payment_service.charge(request.payment_info),
            self.inventory_service.reserve(request.items),
        )
        if not payment_result.success:
            await self.inventory_service.release(request.items)
            raise PaymentFailedError(payment_result.error)
        return ProcessOrderResponse(order_id=payment_result.order_id, status="completed")

See the async example for a complete program.

Stateless vs stateful

Prefer stateless handlers: dependencies on the instance, but no mutable state between invocations. They're thread-safe by default, singleton-friendly, serverless-safe, and easy to test.

When you're tempted to keep state on the handler, put it in an external store instead:

# In-memory state: lost on restart, unsafe across threads/instances
class RateLimitHandler(Handler[RateLimitRequest]):
    def __init__(self, max_requests: int):
        self.request_counts: dict[str, int] = {}   # avoid

# External state: survives restarts, shared across instances
class RateLimitHandler(Handler[RateLimitRequest]):
    def __init__(self, redis_client, max_requests: int):
        self.redis = redis_client
        self.max_requests = max_requests

    def __call__(self, request: RateLimitRequest) -> RateLimitResponse:
        key = f"rate_limit:{request.user_id}"
        current = int(self.redis.get(key) or 0)
        if current >= self.max_requests:
            return RateLimitResponse(allowed=False, remaining=0)
        self.redis.incr(key)
        self.redis.expire(key, 3600)
        return RateLimitResponse(allowed=True, remaining=self.max_requests - current - 1)

Composing operations

Handlers compose by sending further requests through the mediator:

class PlaceOrderHandler(Handler[PlaceOrderRequest]):
    def __init__(self, mediator, database):
        self.mediator = mediator
        self.database = database

    def __call__(self, request: PlaceOrderRequest) -> PlaceOrderResponse:
        inventory = self.mediator.send(CheckInventoryRequest(items=request.items))
        if not inventory.available:
            raise InsufficientInventoryError()

        payment = self.mediator.send(
            ProcessPaymentRequest(amount=request.total, payment_method=request.payment_method)
        )
        order_id = self.database.create_order(items=request.items, payment_id=payment.payment_id)
        self.mediator.send(SendOrderConfirmationRequest(order_id=order_id, email=request.customer_email))
        return PlaceOrderResponse(order_id=order_id)

In async handlers, sub-requests can run in parallel:

class FetchDashboardHandler(Handler[FetchDashboardRequest]):
    def __init__(self, mediator):
        self.mediator = mediator

    async def __call__(self, request: FetchDashboardRequest) -> FetchDashboardResponse:
        user, orders, analytics = await asyncio.gather(
            self.mediator.send(GetUserRequest(user_id=request.user_id)),
            self.mediator.send(GetUserOrdersRequest(user_id=request.user_id)),
            self.mediator.send(GetAnalyticsRequest(user_id=request.user_id)),
        )
        return FetchDashboardResponse(user=user, orders=orders, analytics=analytics)

Testing handlers

A handler is a callable with injected dependencies — test it directly:

def test_create_user_handler():
    mock_db = MockDatabase()
    handler = CreateUserHandler(database=mock_db)

    response = handler(CreateUserRequest(username="testuser", email="test@example.com"))

    assert response.username == "testuser"
    assert response.user_id > 0

Async handlers work the same way with pytest.mark.asyncio:

@pytest.mark.asyncio
async def test_async_handler():
    handler = AsyncFetchHandler(http_client=MockHttpClient())
    response = await handler(FetchDataRequest(url="https://api.example.com/data"))
    assert response.data is not None

More strategies — including faking the mediator for consumers — in the testing guide.

Best practices

Keep handlers focused. Sending a welcome email and updating analytics are their own handlers, dispatched via the mediator — not extra lines in CreateUserHandler.
Inject dependencies; never construct them inside. self.database = DatabaseConnection() makes the handler untestable.
Return rich response objects, not bare primitives — callers shouldn't need a second query for data the handler already had.
Validate early. Check invariants at the top of __call__ before doing work.
Prefer stateless handlers with external state stores.
Type everything. The type system is doing the routing; give it complete information.

Next steps

The mediator — how requests find their handlers
Pipeline behaviors — cross-cutting concerns without touching handlers
Error handling — separating domain errors from framework errors