py_mediatR

Türkçe sürüm: README.tr.md

CQRS / Mediator for Python, inspired by .NET MediatR. Your code sends a small request object; the mediator finds its handler and runs it. The caller never imports or constructs the handler. Zero dependencies; tested on CPython 3.10 to 3.14, including the free-threaded 3.13t/3.14t builds.

Install

pip install py-mediatR              # core (zero dependencies)
pip install "py-mediatR[pydantic]"  # optional pydantic model coercion

Quick start

1. A request and its handler (application/products/list_products.py):

from dataclasses import dataclass
from py_mediatR import IRequest
from domain.product_repository import IProductRepository

@dataclass
class ListProducts(IRequest):
    pass

class ListProductsHandler:                       # found automatically
    def __init__(self, repository: IProductRepository):
        self.repository = repository             # given by the container

    def handle(self, request: ListProducts) -> list[str]:
        return self.repository.list_products()

2. register/register.py - which class serves each contract, how long it lives:

from domain.product_repository import IProductRepository
from infrastructure.memory_product_repository import MemoryProductRepository

def register(container):
    container.register_singleton(IProductRepository, MemoryProductRepository)

3. bootstrap/bootstrap.py - container, registrations and the mediator:

from py_mediatR import Mediator, ServiceContainer
from register import register

def bootstrap() -> Mediator:
    container = ServiceContainer()
    register(container)
    return Mediator(scan_paths=["application"], handler_factory=container,
                    resolve_discovered_from_container=True)

4. Use it anywhere - a console program, a FastAPI/Flask route, a Django view:

mediator = bootstrap()
mediator.send(ListProducts())        # ['Notebook', 'Pencil']

Complete, runnable projects for console, FastAPI, Flask and Django: examples/.

The smallest possible use, without discovery or a container:

from py_mediatR import Mediator, IRequest

class Ping(IRequest):
    pass

class PingHandler:
    def handle(self, req: Ping) -> str:
        return "pong"

mediator = Mediator(auto_discover=False)
mediator.register_handler(Ping, PingHandler())
print(mediator.send(Ping()))  # pong

How a handler is found

There is no naming convention. A class becomes a handler because of the type annotation on its handle() parameter - the class name is irrelevant:

class GetUser(IRequest):
    user_id: str

class ThisNameDoesNotMatter:          # not "GetUserHandler" - still found
    def handle(self, req: GetUser):   # <- the annotation is the contract
        return {"user_id": req.user_id}

Details that bite:

• The annotation must be resolvable at import time. A wrong or unimportable forward reference makes the class invisible rather than raising.
• Exactly one IRequest parameter is allowed; two raises TypeError.
• Extra parameters are fine - cancellation_token is injected by name.
• A handler whose __init__ takes arguments is registered deferred and resolved through handler_factory (your ServiceContainer) at dispatch time.
• One handler per request type; notifications may have many.

Where discovery looks. Mediator() scans the project folder for *.py files. The project folder is found by walking up from the file that creates the Mediator to the nearest folder with requirements.txt, pyproject.toml, setup.py, Pipfile, .git, main.py, app.py or manage.py - so python main.py, uvicorn main:app and test runners find the same project. Pass project_root= to choose it yourself and scan_paths=[...] to narrow it. Folders named venv, .venv, env, site-packages, __pycache__, .git, node_modules, .tox, .nox, .eggs, build, dist, migrations and the .*_cache folders are skipped inside the project. Import errors are logged as warnings on the mediatr.discovery logger unless MEDIATR_DISCOVERY_WARNINGS=0.

Response coercion - the one place naming matters. If a handler returns a dict, py_mediatR coerces it into a response class found by name in the same module: GetUser -> GetUserResponse (or GetUserRequest -> GetUserResponse). An explicit -> GetUserResponse return annotation takes priority. With neither, the dict is returned as-is.

Registering services: ServiceContainer

The second argument creates the object: a class, or a function that takes no argument or takes the container. Handlers and services receive their dependencies through constructor type hints (Optional[T], T | None and Annotated[...] included). A singleton that would capture a scoped service is rejected with DIResolutionError, as in .NET. This check is always on since 6.7.x - there is no validate_scopes switch - so a dependency graph accepted by an older version can be rejected after upgrading.

Frameworks

Every example is a complete project with the same layers and its own README (English and Turkish).

For one scope per HTTP request (a database session shared by the handlers of one request), see Per-request scope below.

Optional features

None of these is needed for the setup above. Each one solves a specific problem; use it only when you have that problem.

Async dispatch

Why: your handlers or behaviors are async def. How: await mediator.send_async(request), await mediator.publish_async(event). Sync and async handlers can be mixed; everything runs on the caller's event loop.

Pipeline behaviors

Why: logging, validation, caching, retries or transactions around many handlers, without repeating that code in each handler. Benefit: one place for each cross-cutting concern. How: every send() runs through the same onion. Behaviors are sorted by order, smallest first, outermost:

send(request)
└─ behavior (order -100)          <- outermost
   └─ behavior (order -50)
      ├─ pre-processor            <- side effects only
      ├─ HANDLER                  <- your business logic
      └─ post-processor           <- sees the response, cannot change it

Eight behaviors are built in. Each reads an attribute off the request; if it is absent the behavior is a no-op, so a request opts in to what it needs:

from dataclasses import dataclass
from py_mediatR import (IRequest, Mediator, LoggingBehavior, ValidationBehavior,
                        CachingBehavior, RetryBehavior, TransactionBehavior,
                        AuthorizationBehavior)

@dataclass
class PlaceOrder(IRequest):
    sku: str
    qty: int = 1
    transactional = True                  # TransactionBehavior engages
    requires_permission = "orders.write"  # AuthorizationBehavior checks this

    def validate(self) -> None:           # ValidationBehavior calls this
        if self.qty < 1:
            raise ValueError("qty must be positive")

mediator = Mediator(behaviors=[
    LoggingBehavior(),
    AuthorizationBehavior(lambda req, perm: current_user.has(perm)),
    ValidationBehavior(),
    CachingBehavior(ttl_seconds=60, max_size=1000),
    RetryBehavior(max_attempts=3, delay=0.1, backoff=2.0, jitter=0.05),
    TransactionBehavior(session_factory=lambda: SessionLocal()),
])

Your own behavior: next_handler() runs the rest of the chain; not calling it short-circuits the handler (that is how CachingBehavior serves a hit).

from py_mediatR import IPipelineBehavior

class AuditBehavior(IPipelineBehavior):
    order = -110                 # lower than LoggingBehavior -> runs outermost
    applies_to = PlaceOrder      # optional; None (default) means every request

    def handle(self, request, next_handler):
        audit.write(f"-> {type(request).__name__}")
        response = next_handler()
        audit.write(f"<- {type(request).__name__}")
        return response

In the async pipeline handle() may be async def, and next_handler() returns an awaitable. Behavior instances are shared across dispatches; mutable state in them needs its own concurrency control.

Per-request scope

Why: the handlers of one HTTP request should share one database session, and it should be closed when the request ends. How: register the session as scoped and dispatch through a scope:

from py_mediatR import scoped_mediator

container.register_scoped(DbSession)
mediator = Mediator(handler_factory=container)
with scoped_mediator(mediator, container) as m:
    m.send(CreateUser(name="Ada"))          # handlers share the scoped session
# scope disposed -> session.close()

FastAPI has a ready dependency for this:

from py_mediatR import make_fastapi_mediator_dependency
get_mediator = make_fastapi_mediator_dependency(mediator, container)

@app.post("/orders")
async def create_order(cmd: CreateOrderDto, m=Depends(get_mediator)):
    return await m.send_async(CreateOrder(**cmd.model_dump()))

Behaviors given as a class (instead of an instance) are created for every dispatch and, through a scope, receive scoped dependencies too.

Pre/post processors

Why: a side effect before or after every handler (audit, metrics). How: both support order and applies_to; a post-processor sees the response but cannot replace it.

from py_mediatR import IRequestPreProcessor, IRequestPostProcessor

class AuditPre(IRequestPreProcessor):
    def process(self, request):                 # may be async def
        audit.write(type(request).__name__)

class MetricsPost(IRequestPostProcessor):
    def process(self, request, response):       # may be async def
        metrics.increment(type(request).__name__)

Exception handlers and actions

Why: turn a known error into a fallback response, or observe every error.

from py_mediatR import IExceptionHandler, IExceptionAction

class InvoiceNotFound(IExceptionHandler):
    exception_type = KeyError
    applies_to = FindInvoice
    def handle(self, request, exc, state):
        state.set_handled(FindInvoiceResponse(invoice_id=request.invoice_id, found=False))

class AlertAction(IExceptionAction):
    exception_type = Exception
    def execute(self, request, exc):
        alerting.notify(f"{type(request).__name__} failed: {exc}")

Both filter on exception_type, applies_to and order. All matching actions run; the first matching handler that marks the error handled wins. A two-parameter handle(request, exc) that returns a non-None value also counts as handled.

Validators

Why: validation that needs dependencies, outside the request class (FluentValidation style).

from py_mediatR import IValidator, ValidationBehavior

class CreateUserValidator(IValidator):
    applies_to = CreateUser          # or a tuple of types; None = all
    def validate(self, request):
        if "@" not in request.email:
            raise ValueError(f"invalid e-mail: {request.email}")

mediator = Mediator(behaviors=[ValidationBehavior(validators=[CreateUserValidator()])])

Matching validators run in order, then the request's own validate(). Async validators are awaited in send_async() and rejected in send().

Notifications

Why: one event, several independent reactions (send an e-mail, update a report...).

from py_mediatR import PublishStrategy

mediator.publish(UserRegistered(user_id="U-1"))                       # SEQUENTIAL
await mediator.publish_async(evt, strategy=PublishStrategy.PARALLEL_WHENALL)
await mediator.publish_async(evt, strategy=PublishStrategy.PARALLEL_NOWAIT)

Subscriber errors propagate by default; Mediator(swallow_notification_errors=True) logs them instead. With polymorphic_publish=True, publishing a subclass also triggers base-type subscribers.

Streaming

Why: produce many items one by one (IAsyncEnumerable<T> in .NET).

from py_mediatR import IStreamRequest, IStreamPipelineBehavior

class StreamOrders(IStreamRequest):
    count: int = 100

class StreamOrdersHandler:
    async def handle(self, req: StreamOrders):
        for i in range(req.count):
            yield {"seq": i}

async for order in mediator.create_stream(StreamOrders(count=10)):
    ...

IStreamPipelineBehavior wraps the generator like a pipeline behavior.

Cancellation

Why: stop a slow request (timeout, user cancelled).

from py_mediatR import CancellationTokenSource, OperationCancelledError

cts = CancellationTokenSource(cancel_after=2.0)   # timeout
try:
    result = await mediator.send_async(SlowQuery(), cancellation_token=cts.token)
except OperationCancelledError:
    ...

A handler that declares a cancellation_token parameter receives the token. Linked tokens (CancellationTokenSource.create_linked(...)) are supported.

See the call chain

Why: "why didn't my handler run?" - a mediator hides who calls what.

with mediator.trace() as flow:
    mediator.send(SearchProducts(keyword="laptop"))
    mediator.send(SearchProducts(keyword="laptop"))   # served from cache
print(flow.render())

send(SearchProducts)   (0.35 ms)
└─ behavior: LoggingBehavior   (0.29 ms)
   └─ behavior: CachingBehavior   [cache miss]   (0.20 ms)
      └─ HANDLER: SearchProductsHandler   (0.07 ms)
send(SearchProducts)   (0.14 ms)
└─ behavior: LoggingBehavior   (0.11 ms)
   └─ behavior: CachingBehavior   [CACHE HIT - handler NOT called]

Tracing is off by default. flow.steps() and flow.find(label) expose the same data for assertions; render(unicode=False) forces ASCII.

Strict mode

Why: turn silent fallbacks into errors while developing.

mediator = Mediator(strict_async=True, strict_discovery=True, strict_response=True)

• strict_async - an awaitable returned through synchronous send()/publish() raises TypeError naming the component (otherwise a WARNING on mediatr.dispatch).
• strict_discovery - all duplicate-handler and import problems of the scan raise one DiscoveryError (a ValueError); otherwise they are WARNING logs.
• strict_response - a failing dict -> dataclass/Pydantic response conversion re-raises (otherwise a WARNING on mediatr.response and the raw result).

Explicit registration

Why: large codebases where scanning is not wanted.

mediator = Mediator(auto_discover=False)
mediator.register_handler(GetUser, GetUserHandler())

or mark classes and let discovery collect only them:

from py_mediatR import handler

@handler
class GetUserHandler:
    def handle(self, req: GetUser): ...

Notes

• Transactions: a failing rollback()/close() is logged and attached to the original exception (add_note), or raised as TransactionCleanupError with raise_on_cleanup_failure=True.
• Sync-over-async bridge: waits at most 30 s (MEDIATR_SYNC_BRIDGE_TIMEOUT, <=0 disables) and raises SyncBridgeTimeoutError instead of freezing.
• Debug output: MEDIATR_DEBUG=1 prints discovery details; flow.print() writes to stdout. Library diagnostics otherwise use logging.
• Known limits: behavior instances are shared across dispatches and scope clones; stream cancellation and early-close cleanup through nested stream wrappers are limited; strict async checks cover send/publish, not streams. Scope cleanup failures emit both a RuntimeWarning and a WARNING log.
• Imports: import py_mediatR (also import py_mediatr). Everything in __all__ is re-exported at the top level; the submodules (contracts, mediator, di, discovery, behaviors, cancellation, tracing, coercion) are an implementation detail.
• Changes per version: CHANGELOG.md.

Test quality

The suite has 444 tests and runs on CPython 3.10, 3.11, 3.14 and the free-threaded 3.14t build (GIL disabled). The whole package was mutation tested with cosmic-ray: 2406 mutants, 100% of those that can change behaviour killed (92.5% of all; the rest are proven equivalent, each class with its reason). Report: docs/MUTATION_REPORT.md.

Examples

• examples/
  • four small, self-contained projects (console, FastAPI, Flask, Django). Start here.
• examples/advanced/
  • a bigger e-commerce demo that uses every feature, with a ten-step console walkthrough that prints each call chain.

The examples are not shipped in the wheel; they live in the repository.

License

MIT - Ayhan Öztemel