Error handling

PyMediate does not translate exceptions raised by request handlers or pipeline behaviors. send() propagates them unchanged, so the application's entry point can decide how to report a failure over HTTP, a command-line interface, or another transport.

It helps to distinguish three sources:

• Application errors describe a failed operation, such as a missing product or insufficient stock.
• Transport errors describe how an entry point reports that failure, such as an HTTP status and response body.
• PyMediate errors report invalid declarations or missing dispatch configuration.

Define application errors without a transport

An application error can carry the domain values a caller needs without importing a web framework:

class ShopError(Exception):
    """Base class for shop operation failures."""


class ProductNotFoundError(ShopError):
    def __init__(self, product: str) -> None:
        self.product = product
        super().__init__(f"product not found: {product}")


class OutOfStockError(ShopError):
    def __init__(self, product: str, requested: int, available: int) -> None:
        self.product = product
        self.requested = requested
        self.available = available
        super().__init__(
            f"not enough {product}: requested {requested}, available {available}",
        )

A handler can raise these errors as part of its operation:

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

    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        available = await self._inventory.available(request.item)
        if available is None:
            raise ProductNotFoundError(request.item)
        if available < request.quantity:
            raise OutOfStockError(request.item, request.quantity, available)

        order_id = await self._orders.add(request)
        return OrderReceipt(
            order_id=order_id,
            summary=f"{request.quantity} × {request.item}",
        )

Avoid adding an HTTP status or framework exception to these classes solely for one entry point. That detail would make a command-line command or scheduled job depend on an HTTP convention it does not use.

Translate errors at the entry point

Web frameworks provide central exception handlers. For example, FastAPI can map the same application errors to two HTTP responses:

from fastapi import FastAPI, Request as HTTPRequest
from fastapi.responses import JSONResponse

app = FastAPI()


@app.exception_handler(ProductNotFoundError)
async def product_not_found(
    request: HTTPRequest,
    error: ProductNotFoundError,
) -> JSONResponse:
    return JSONResponse(
        status_code=404,
        content={"error": str(error), "product": error.product},
    )


@app.exception_handler(OutOfStockError)
async def out_of_stock(
    request: HTTPRequest,
    error: OutOfStockError,
) -> JSONResponse:
    return JSONResponse(
        status_code=409,
        content={"error": str(error), "available": error.available},
    )

A command-line entry point can catch the same classes and choose an exit code and text output. The handler does not change between those entry points.

A pipeline behavior can implement a transport-neutral error policy, such as recording a failure and re-raising it. Translating application errors into HTTPException inside a behavior makes that mediator configuration specific to HTTP. A framework-level exception handler is usually clearer when the mapping is only for that framework.

Understand definition-time errors

Request, notification, and stream handlers are validated when Python defines their classes. These errors normally stop module import and indicate a declaration that needs to be corrected:

These six exceptions inherit from PyMediateError. ResponseTypeMismatchError checks the return annotation; PyMediate does not execute the handler or inspect a later response value during class definition.

Static type checkers catch some of the same mistakes. Definition-time validation also runs when unchecked Python imports the module. See type safety for the boundary between those checks.

Understand wiring-time errors

Two errors report a service collection or pipeline that cannot work, and both raise while the application is being wired rather than during a dispatch:

Both inherit from PyMediateError, and both fail before the mediator can be used, so a misconfigured wiring never reaches a request.

Understand dispatch-time errors

Two public errors describe missing resolution configuration:

HandlerNotFoundError

send() or stream() raises HandlerNotFoundError when the exact request type has no declared handler class. Common causes include a missing handler definition or an application module that was not imported during startup.

HandlerNotFoundError inherits from PyMediateError.

ServiceNotFoundError

Once a handler class is known, the mediator asks the service provider for an instance of that exact class. If the provider does not contain one, it raises ServiceNotFoundError. Notification publication raises the same error when a declared subscriber has no service instance; all subscriber instances are resolved before any subscriber runs.

ServiceNotFoundError inherits from KeyError — the miss surfaces through the provider[type] subscript, so except KeyError catches it — not from PyMediateError. Catching PyMediateError therefore does not catch missing services.

Both errors normally indicate application setup rather than invalid user input. An entry point can log them as configuration failures while allowing application errors to use their own mappings.

The optional Dependency Injector adapter can also raise TypeError for an unsupported or asynchronous provider and ValueError for a nested-container cycle. Those errors occur while the adapter indexes or resolves the container; see dependency injection.

Know when handler errors appear

Each dispatch form propagates user-code failures at a different point:

For asynchronous publication, ordinary exceptions produce an ExceptionGroup. Other collected failures derived directly from BaseException produce a BaseExceptionGroup. KeyboardInterrupt and SystemExit are handled specially by Python: they propagate directly and can cancel unfinished subscribers. The synchronous mediator groups ordinary Exception instances; a direct BaseException is not collected.

Use except* to handle selected failures from a grouped publish:

try:
    await mediator.publish(OrderPlaced(order_id=42, item="tea"))
except* ConnectionError as group:
    for error in group.exceptions:
        print(f"subscriber unavailable: {error}")

The notifications guide covers subscriber resolution, concurrency, and grouping in more detail.

Continue

• Map application errors in a FastAPI entry point.
• Diagnose individual library exceptions in troubleshooting.
• Look up the exact error hierarchy and signatures.