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.
HexCore’s middleware pipeline sits between the bus and every handler, giving you a clean place to inject cross-cutting concerns — logging, transactions, retries, validation, or any custom logic — without coupling those concerns to individual handlers. The pipeline follows the Chain of Responsibility pattern: each middleware receives the message and a next_handler callable that invokes the next link in the chain, ultimately reaching the domain handler.
AbstractMiddleware
Every middleware inherits from hexcore.domain.cqrs.middleware.AbstractMiddleware and implements a single async method:
# hexcore/domain/cqrs/middleware.py
import abc
import typing as t
NextHandler = t.Callable[[t.Any], t.Awaitable[t.Any]]
class AbstractMiddleware(abc.ABC):
@abc.abstractmethod
async def handle(
self,
message: t.Any,
next_handler: NextHandler,
) -> t.Any:
"""
Process the message and delegate to the next handler.
Args:
message: The Command, Query, or Event being processed.
next_handler: Callable that invokes the next middleware or the final handler.
Returns:
The result of execution (may be None for commands/events).
"""
raise NotImplementedError
The Command, Query, or DomainEvent being processed. The middleware is generic — the same class can wrap any message type.
An async callable (message) -> result. Calling it invokes the next middleware in the chain, or the final domain handler if this is the last middleware. You must await it to propagate the message.
How MiddlewarePipeline builds the chain
MiddlewarePipeline constructs the chain lazily at execution time by wrapping middlewares from right to left:
[LoggingMiddleware] → [RetryMiddleware] → [TransactionMiddleware] → [handler.handle]
# hexcore/application/cqrs/pipeline.py (simplified)
from hexcore.domain.cqrs.middleware import IMiddleware, NextHandler
import typing as t
class MiddlewarePipeline:
def __init__(self, middlewares: t.Sequence[IMiddleware] | None = None) -> None:
self._middlewares: list[IMiddleware] = list(middlewares or [])
def add(self, middleware: IMiddleware) -> "MiddlewarePipeline":
"""Fluent API: add one middleware and return self."""
self._middlewares.append(middleware)
return self
def add_many(self, middlewares) -> "MiddlewarePipeline":
self._middlewares.extend(middlewares)
return self
async def execute(self, message, final_handler: NextHandler):
chain = final_handler
for middleware in reversed(self._middlewares):
chain = self._wrap(middleware, chain)
return await chain(message)
Built-in middlewares
HexCore ships four production-ready middlewares in hexcore.infrastructure.cqrs.middlewares:
TransactionMiddleware
Wraps command execution in a Unit of Work transaction. Commits on success and rolls back on any exception. This is the default middleware on the command bus when using CQRSConfig.
from hexcore.infrastructure.cqrs.middlewares import TransactionMiddleware
# Default: auto-detects SQL or NoSQL UoW from ServerConfig
middleware = TransactionMiddleware()
# Custom factory: inject your own UoW
from hexcore.infrastructure.uow import SqlAlchemyUnitOfWork
middleware = TransactionMiddleware(
uow_factory=lambda: SqlAlchemyUnitOfWork(session=get_session())
)
LoggingMiddleware
Logs the message type, execution time in milliseconds, and any exceptions.
from hexcore.infrastructure.cqrs.middlewares import LoggingMiddleware
import logging
middleware = LoggingMiddleware(
logger=logging.getLogger("myapp.cqrs"),
log_level=logging.DEBUG,
)
RetryMiddleware
Retries the handler on failure with exponential backoff. By default retries all exceptions up to 3 times.
from hexcore.infrastructure.cqrs.middlewares import RetryMiddleware
import asyncpg
middleware = RetryMiddleware(
max_retries=3,
base_delay=0.1, # seconds
retryable_exceptions=(asyncpg.PostgresConnectionFailureError,),
)
ValidationMiddleware
Re-validates the Pydantic model before passing it to the handler. Useful when commands arrive from external sources (e.g. deserialized from a queue).
from hexcore.infrastructure.cqrs.middlewares import ValidationMiddleware
middleware = ValidationMiddleware()
Writing a custom middleware
Any class that inherits AbstractMiddleware and implements handle() is a valid middleware.
import time
import logging
from hexcore.domain.cqrs.middleware import AbstractMiddleware, NextHandler
logger = logging.getLogger("myapp")
class TimingMiddleware(AbstractMiddleware):
async def handle(self, message, next_handler: NextHandler):
start = time.perf_counter()
try:
result = await next_handler(message)
elapsed_ms = (time.perf_counter() - start) * 1000
logger.info(
"[Timing] %s completed in %.2fms",
type(message).__name__,
elapsed_ms,
)
return result
except Exception as exc:
elapsed_ms = (time.perf_counter() - start) * 1000
logger.error(
"[Timing] %s failed after %.2fms: %s",
type(message).__name__,
elapsed_ms,
exc,
)
raise
Chaining multiple middlewares
from hexcore.infrastructure.cqrs.middlewares import (
LoggingMiddleware,
RetryMiddleware,
TransactionMiddleware,
)
from myapp.middlewares import TimingMiddleware
logging_mw = LoggingMiddleware()
timing_mw = TimingMiddleware()
retry_mw = RetryMiddleware(max_retries=2)
transaction_mw = TransactionMiddleware()
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
pipeline = (
MiddlewarePipeline()
.add(logging_mw) # outermost — runs first, logs start/end
.add(timing_mw) # second — measures execution time
.add(retry_mw) # third — retries on failure
.add(transaction_mw) # innermost — wraps handler in a UoW transaction
)
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
from hexcore.application.cqrs.registry import HandlerRegistry
bus = InMemoryCommandBus(registry=registry, pipeline=pipeline)
Complete example
import asyncio
import logging
from hexcore.domain.cqrs.middleware import AbstractMiddleware, NextHandler
from hexcore.domain.cqrs.commands import Command
from hexcore.domain.cqrs.handlers import AbstractCommandHandler
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
logging.basicConfig(level=logging.DEBUG)
# ── Custom middleware ─────────────────────────────────────────────
class AuditMiddleware(AbstractMiddleware):
async def handle(self, message, next_handler: NextHandler):
print(f"[Audit] Before: {type(message).__name__} id={getattr(message, 'command_id', '?')}")
result = await next_handler(message)
print(f"[Audit] After: {type(message).__name__} result={result}")
return result
# ── Command and handler ───────────────────────────────────────────
class GreetCommand(Command):
name: str
class GreetHandler(AbstractCommandHandler[GreetCommand, str]):
async def handle(self, command: GreetCommand) -> str:
return f"Hello, {command.name}!"
# ── Wiring ────────────────────────────────────────────────────────
from hexcore.infrastructure.cqrs.middlewares import LoggingMiddleware
pipeline = (
MiddlewarePipeline()
.add(LoggingMiddleware())
.add(AuditMiddleware())
)
registry = HandlerRegistry()
registry.register_command_handler(GreetCommand, GreetHandler())
bus = InMemoryCommandBus(registry=registry, pipeline=pipeline)
# ── Dispatch ──────────────────────────────────────────────────────
async def main():
result = await bus.dispatch(GreetCommand(name="HexCore"))
print(result)
asyncio.run(main())
Declarative middleware via CQRSConfig
In production you can configure middlewares by dotted import path so the CQRSFactory resolves and instantiates them automatically:
from hexcore.application.cqrs.config import CQRSConfig, BusConfig
config = CQRSConfig(
command_bus=BusConfig(
middlewares=[
"hexcore.infrastructure.cqrs.middlewares.LoggingMiddleware",
"myapp.middlewares.AuditMiddleware",
"hexcore.infrastructure.cqrs.middlewares.RetryMiddleware",
"hexcore.infrastructure.cqrs.middlewares.TransactionMiddleware",
],
),
)
Middlewares specified in BusConfig.middlewares must have a zero-argument constructor. If a middleware needs configuration (e.g. custom logger, retry count), instantiate it directly and pass the MiddlewarePipeline instance to InMemoryCommandBus rather than using the declarative string approach.
Execution order cheat-sheet
Given .add(A).add(B).add(C):
Request → A.handle → B.handle → C.handle → handler.handle
Response ← A.handle ← B.handle ← C.handle ← handler.handle
A wraps the entire execution, making it the best place for global logging and audit trails. Middleware C is closest to the handler, making it the best place for transactions and retries.