Middlewares

A middleware wraps the execution of a handler. Use it to run logic before and after the handler processes the message, or to deal with its exceptions. Middlewares are added to a bus with add_middlewares:

bus = new_command_bus()
bus.add_middlewares(timing_middleware, transaction_middleware)

For commands and queries, the middleware stack wraps the single handler once. For events, the stack is applied around each handler independently, so a middleware sees one invocation per event handler.

The same middlewares can be attached to a command pipeline and to a Pump.

Generator middlewares

The simplest form is an async generator that receives the message and yields once, where the handler should run:

import time


async def timing_middleware(message):
    start = time.perf_counter()
    yield
    print(f"Execution time: {time.perf_counter() - start}s")

The yield behaves like a try/finally around the handler call. An exception raised by the handler is thrown at the yield, so you can catch it there:

async def logging_middleware(message):
    try:
        yield
    except Exception:
        logger.exception("Failed to handle %r", message)
        raise

result = yield receives the handler's return value, but only for inspection: a generator middleware cannot replace it. This keeps the message and the result read-only by default. If the middleware swallows an exception instead of re-raising it, dispatch returns the NotImplemented sentinel.

After an exception, the middleware may yield again to call the handler one more time. This is how RetryMiddleware is written. Yielding again after a successful call raises cq.exceptions.MiddlewareError.

The return annotation for these middlewares is MiddlewareResult[T], where T is the handler's return type:

from cq import MiddlewareResult
from typing import Any


async def timing_middleware(message: Any) -> MiddlewareResult[Any]:
    ...
    yield

Classic middlewares

When you need to replace the return value, or to skip the handler and return something else instead, write a "classic" middleware. It takes call_next as its first argument and returns the value it wants to expose to the caller:

async def cache_middleware(call_next, query):
    if (cached := cache.get(query)) is not None:
        return cached

    result = await call_next(query)
    cache.set(query, result)
    return result

Both styles can be mixed freely on the same bus. Middleware is the type alias that covers both.

Finding the handler

resolve_handler_source(call_next) returns the handler class behind call_next, so a classic middleware knows which handler is about to run, typically to log or trace it:

from cq import resolve_handler_source


async def logging_middleware(call_next, message):
    handler_type = resolve_handler_source(call_next)
    logger.info("%s handles %r", handler_type.__name__, message)
    return await call_next(message)

The class also gives access to metadata declared on the handler, such as a class attribute or a marker set by your own decorator.

It only works on a bus. On a Pump or a pipeline, there is no handler behind call_next, and it raises an AttributeError.

Class-based middlewares

Both styles can also be written as a class with an async __call__ method, which is convenient when the middleware needs its own dependencies:

from dataclasses import dataclass


@dataclass
class TimingMiddleware:
    metrics: MetricsService

    async def __call__(self, message):
        start = time.perf_counter()
        yield
        await self.metrics.record(time.perf_counter() - start)


@dataclass
class CacheMiddleware:
    cache: Cache

    async def __call__(self, call_next, query): ...

If you build these classes through your DI container (for example as parameters of the bus factory), you get the same constructor injection as for handlers.

Execution order

Middlewares nest around the handler like the layers of an onion. Within one call, they run in the order you list them: the first argument is the outermost layer, so it starts first and finishes last.

bus.add_middlewares(outer, inner)
# outer -> inner -> handler -> inner -> outer

Each subsequent call wraps whatever is already registered, so the middlewares added last end up outermost:

bus.add_middlewares(a, b)  # a -> b -> handler
bus.add_middlewares(c)  # c -> a -> b -> handler

The rule also covers the command scope that new_command_bus() installs before you add anything. Your own middlewares sit outside of it: they wrap the dispatch of every related event, but they cannot inject a dependency that only exists inside the scope, such as RelatedEvents.

Built-in middlewares

RetryMiddleware

cq.middlewares.retry.RetryMiddleware calls the handler again when it raises one of the configured exception types:

from cq.middlewares.retry import RetryMiddleware

bus.add_middlewares(RetryMiddleware(retry=3, delay=0.5, exceptions=(TimeoutError,)))

• retry: total number of attempts, including the first one. With retry=3, the handler runs at most three times. Must be at least 1.
• delay: seconds to wait between attempts. Defaults to 0.
• exceptions: the exception types that trigger a retry. Defaults to (Exception,).

If every attempt fails, the last exception is re-raised.

CaptureExceptionMiddleware

cq.middlewares.exc.CaptureExceptionMiddleware catches exceptions raised downstream and forwards them to a callback. Use it to log, report, or push errors to an external sink:

from cq.middlewares.exc import CaptureExceptionMiddleware


async def report(exception, message):
    sentry_sdk.capture_exception(exception)


bus.add_middlewares(CaptureExceptionMiddleware(report, reraise=True))

• on_error: an async callback invoked with the captured exception followed by the arguments the handler received (the message). Use CaptureExceptionMiddleware.sync(...) if your callback is synchronous.
• exceptions: the exception types to capture. Defaults to (Exception,).
• reraise: whether to re-raise the exception after the callback returns. Defaults to False, in which case the exception is swallowed.

on_error is meant for side effects only and must not raise. If it does, its own exception propagates in place of the original one.

Context manager middlewares

cq.middlewares.contextlib turns an existing context manager into a middleware that wraps every dispatch:

• ContextManagerMiddleware(context) for a synchronous context manager;
• AsyncContextManagerMiddleware(context) for an asynchronous one.

from contextlib import suppress
from cq.middlewares.contextlib import ContextManagerMiddleware

bus.add_middlewares(ContextManagerMiddleware(suppress(LookupError)))

The same context manager instance is entered on every dispatch, so it must be reusable. Context managers built with @contextmanager are single-use and do not fit here.