Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt

Use this file to discover all available pages before exploring further.

Middleware lets you intercept every command, query, or event that flows through a bus — before and after the handler executes — without modifying the handler itself. HexCore uses the Chain of Responsibility pattern: each middleware in the pipeline receives the message and a next_handler callable that invokes the rest of the chain. MiddlewarePipeline assembles the chain and executes it; AbstractMiddleware is the contract every middleware must implement.

​
NextHandler type alias

from hexcore.domain.cqrs.middleware import NextHandler

NextHandler = Callable[[Any], Awaitable[Any]]
A type alias for the callable that advances to the next step in the middleware chain. Always await it and propagate its return value so downstream middlewares and the final handler can complete normally.

​
AbstractMiddleware

from hexcore.domain.cqrs.middleware import AbstractMiddleware

class AbstractMiddleware(abc.ABC):
    @abstractmethod
    async def handle(self, message: Any, next_handler: NextHandler) -> Any: ...
The abstract contract every middleware must implement. Subclass this and override handle to add cross-cutting behaviour (logging, timing, authentication, transaction management, etc.).

​
Abstract method

​
handle

@abstractmethod
async def handle(self, message: Any, next_handler: NextHandler) -> Any
Processes the message and must call (and await) next_handler(message) to continue the chain. You can execute logic before the call (pre-processing), after the call (post-processing), or wrap both in a try/except for error handling.
​
message
Any
required
The Command, Query, or DomainEvent being processed. The exact type depends on which bus the middleware is attached to.
​
next_handler
NextHandler
required
A callable that invokes the next middleware in the chain, or the final handler if this is the last middleware. You must await it and return its result unless you deliberately want to short-circuit the chain.
​
result
Any
The result returned by calling next_handler(message). Pass this through unchanged unless your middleware intentionally transforms the result.

​
IMiddleware alias

IMiddleware = AbstractMiddleware
A backward-compatibility alias. Existing code importing IMiddleware continues to work unchanged.

​
MiddlewarePipeline

from hexcore.application.cqrs.pipeline import MiddlewarePipeline
Builds and executes a chain of middlewares. The pipeline is constructed lazily: calling execute() wraps the middlewares around the final handler at runtime, with the first middleware added being the outermost wrapper (it runs first before any others). 
[MW1] → [MW2] → ... → [MWn] → [handler.handle]

​
Constructor

class MiddlewarePipeline:
    def __init__(
        self,
        middlewares: Sequence[IMiddleware] | None = None,
    ) -> None: ...
​
middlewares
Sequence[IMiddleware] | None
default:"None"
An optional initial list of middleware instances. Equivalent to calling add() for each element after construction.

​
add

def add(self, middleware: IMiddleware) -> MiddlewarePipeline
Appends a single middleware to the end of the pipeline. Returns self for fluent chaining.
​
middleware
IMiddleware
required
The middleware instance to append.
​
pipeline
MiddlewarePipeline
Returns self to allow: pipeline.add(LoggingMiddleware()).add(TimingMiddleware()).

​
add_many

def add_many(self, middlewares: Iterable[IMiddleware]) -> MiddlewarePipeline
Appends multiple middleware instances in a single call.
​
middlewares
Iterable[IMiddleware]
required
An iterable of middleware instances to append.
​
pipeline
MiddlewarePipeline
Returns self for fluent chaining.

​
execute

async def execute(
    self,
    message: Any,
    final_handler: NextHandler,
) -> Any
Executes the full middleware chain for a single message. Builds the chain by wrapping middlewares from right to left (so the first middleware is outermost), then awaits the resulting chain with message.
​
message
Any
required
The command, query, or event to process.
​
final_handler
NextHandler
required
The async callable that represents the actual handler (e.g., handler.handle). Called last after all middlewares have run.
​
result
Any
The value returned by final_handler, propagated back through the chain.

​
Usage examples

import logging
from hexcore.domain.cqrs.middleware import AbstractMiddleware, NextHandler

logger = logging.getLogger("app.cqrs")

class LoggingMiddleware(AbstractMiddleware):
    async def handle(self, message: Any, next_handler: NextHandler) -> Any:
        msg_type = type(message).__name__
        logger.info("→ Handling %s", msg_type)
        try:
            result = await next_handler(message)
            logger.info("✓ Completed %s", msg_type)
            return result
        except Exception as exc:
            logger.error("✗ Failed %s: %s", msg_type, exc)
            raise
The order middlewares are added is the order they wrap the chain. MW1.add(MW2) means MW1 runs first (outermost), then MW2, then the handler. This mirrors the behaviour of popular frameworks like Django or Express middleware stacks.
Always await next_handler(message) and return its result. Forgetting to call next_handler silently short-circuits the chain: the final handler and all subsequent middlewares will never execute.

Build docs developers (and LLMs) love

Get started for freeTalk to us