Python 3.12+ PEP 695 generics MIT

Type-safe request dispatch for modern Python

PyMediate routes typed requests to their handlers through one mediator — response types inferred end to end, first-class async, zero runtime dependencies.

Get started Documentation

One `send()`, fully inferred

Declare the response type once, on the request. From there the mediator, the handler signature, and every call site agree — and mypy enforces it. The async API is a structural mirror: switch the import, add await, done.

Learn the core concepts

app.py

from dataclasses import dataclass
from pymediate import Request, Handler, Mediator, Services

@dataclass
class UserCreated:
    user_id: int
    username: str

@dataclass
class CreateUser(Request[UserCreated]):
    username: str
    email: str

class CreateUserHandler(Handler[CreateUser]):
    def __call__(self, request: CreateUser) -> UserCreated:
        return UserCreated(user_id=1, username=request.username)

services = Services()
services.add(CreateUserHandler())
mediator = Mediator(services.provider())

response = mediator.send(CreateUser(username="alice", email="alice@example.com"))
# response is UserCreated — inferred from the request, checked by mypy

app.py

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

@dataclass
class UserCreated:
    user_id: int
    username: str

@dataclass
class CreateUser(Request[UserCreated]):
    username: str
    email: str

class CreateUserHandler(Handler[CreateUser]):
    async def __call__(self, request: CreateUser) -> UserCreated:
        user_id = await user_repository.save(request.username, request.email)
        return UserCreated(user_id=user_id, username=request.username)

mediator = Mediator(Services().add(CreateUserHandler()).provider())

response = await mediator.send(CreateUser(username="alice", email="alice@example.com"))

behaviors.py

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

class LoggingBehavior(PipelineBehavior[Request]):
    """Applies to every request — before and after its handler runs."""

    def __call__(self, request: Request, next: Callable[[], Any]) -> Any:
        print(f"handling {type(request).__name__}")
        response = next()
        print(f"returning {type(response).__name__}")
        return response

class AuditCreateUser(PipelineBehavior[CreateUser]):
    """Selective: only wraps CreateUser requests."""

    def __call__(self, request: CreateUser, next: Callable[[], Any]) -> Any:
        audit_log.record(request.email)
        return next()

services.add(LoggingBehavior())
services.add(AuditCreateUser())

Small surface, sharp edges filed off

A handful of concepts — requests, handlers, behaviors, one mediator — designed to stay out of your way.

Typed end to end

send() returns exactly what the request declares — response types are inferred from Request[T], validated by mypy --strict, no casts.

Zero dependencies

The core is pure Python 3.12+ using PEP 695 generics. One optional extra when you want a DI container, nothing else.

Async mirror

pymediate.aio mirrors the sync API structurally — same classes, same semantics, await where it matters.

Pipeline behaviors

Wrap every handler — or just some — with logging, validation, caching, or transactions, without touching handler code.

Fails at definition time

Handler signatures are validated when the class is defined, so wiring mistakes surface at import — not in production.

DI, your way

Use the built-in Services registry, or put a dependency-injector container behind the ServiceProvider protocol.

Why a mediator?

Decouple callers from handlers

The code that sends CreateUser never imports the code that handles it. Features stay independent, and new handlers slot in without changing existing code.

CQRS by construction

Commands and queries are just request types. Separating writes from reads becomes a naming convention, not framework machinery.

Trivially testable

Handlers are plain callables — call them directly in tests. Consumers depend only on the mediator, so faking it is one line.

Ship your first handler in five minutes

Install the package, define a request, write a handler — the quick start walks you through the rest.

Quick start API reference