Quick start

Define a request, write a handler, and send it through a mediator — a working feature in five minutes.

This page builds a complete user-creation feature: a request and response, a handler, and a mediator to connect them.

The complete picture

A full, runnable feature in one file:

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

# 1. Define the response
@dataclass
class UserCreated:
    user_id: int
    username: str
    email: str

# 2. Define the request
@dataclass
class CreateUser(Request[UserCreated]):
    username: str
    email: str

# 3. Write the handler
class CreateUserHandler(Handler[CreateUser]):
    def __init__(self):
        self.next_id = 1
        self.users = {}

    def __call__(self, request: CreateUser) -> UserCreated:
        user_id = self.next_id
        self.next_id += 1
        self.users[user_id] = {"username": request.username, "email": request.email}
        return UserCreated(user_id=user_id, username=request.username, email=request.email)

# 4. Set up the mediator
services = Services()
services.add(CreateUserHandler())
mediator = Mediator(services.provider())

# 5. Send a request
response = mediator.send(CreateUser(username="alice", email="alice@example.com"))
print(f"Created user {response.username} with ID {response.user_id}")
# Output: Created user alice with ID 1

The rest of this page walks through the five numbered pieces.

The type parameter tells PyMediate — and mypy — what response this request produces. That single declaration drives type inference everywhere: the handler's return type and the result of mediator.send() both follow from it.

Write the handler

A handler subclasses Handler[CreateUser] and implements __call__. The type parameter is how PyMediate knows which requests it handles — no naming conventions, no registration strings:

from pymediate import Handler

class CreateUserHandler(Handler[CreateUser]):
    def __init__(self):
        self.next_id = 1
        self.users = {}

    def __call__(self, request: CreateUser) -> UserCreated:
        user_id = self.next_id
        self.next_id += 1
        self.users[user_id] = {"username": request.username, "email": request.email}
        return UserCreated(user_id=user_id, username=request.username, email=request.email)

Set up the mediator

from pymediate import Mediator, Services

services = Services()
services.add(CreateUserHandler())

mediator = Mediator(services.provider())

Send a request

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

print(f"Created user {response.username} with ID {response.user_id}")
# Output: Created user alice with ID 1

response is typed as UserCreated. Your editor autocompletes its fields, and mypy verifies every use.

Type safety in action

PyMediate validates handler signatures when the class is defined, so a wiring mistake fails at import time:

class WrongHandler(Handler[CreateUser]):
    def __call__(self, request: CreateUser) -> str:  # wrong return type
        return "oops"

# TypeError: WrongHandler.__call__ must return UserCreated, got str

Async support

The async API in pymediate.aio is a structural mirror of the sync one — same classes, same flow:

from pymediate import Services
from pymediate.aio import Handler, Mediator

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

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

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

See the async example for a complete program.

Next steps

Learn the core concepts behind requests, handlers, and the mediator.
Add cross-cutting logic with pipeline behaviors.
Wire handlers through a container with dependency injection.

The complete picture

Step by step

Define the response

Define the request