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.

HexCore’s CQRS subsystem connects commands, queries, and events to their handlers through a trio of buses backed by a central HandlerRegistry. This page walks through every step needed to get a fully wired CQRS layer running inside a HexCore application — from creating the registry and registering handlers, to configuring buses declaratively with CQRSConfig, and exposing them as FastAPI dependencies.

​
Core building blocks

HandlerRegistry

Maps every Command and Query type to its handler. Supports eager instance registration and lazy factory registration for DI containers.

MiddlewarePipeline

Chain of AbstractMiddleware instances that wrap every message before it reaches the handler. Built and executed lazily per message.

InMemoryCommandBus

Resolves the correct handler from the registry, runs the pipeline, and returns the result. Supports Smart Routing to background queues.

InMemoryQueryBus

Always synchronous. Resolves a query handler from the registry and returns the typed result directly.

​
Step-by-step wiring

1
Install HexCore
2
pip install hexcore
3
Create a HandlerRegistry and register handlers
4
HandlerRegistry is the single source of truth for which handler processes each message type. Both eager instances and lazy factory callables are supported — factories are resolved once and then cached.
5
from hexcore.application.cqrs.registry import HandlerRegistry
from myapp.commands import CreateOrderCommand
from myapp.handlers import CreateOrderHandler
from myapp.queries import GetOrderQuery
from myapp.query_handlers import GetOrderQueryHandler

registry = HandlerRegistry()

# Eager registration — handler instantiated right now
registry.register_command_handler(CreateOrderCommand, CreateOrderHandler(order_repo))

# Lazy factory — handler created on first dispatch
registry.register_query_handler(
    GetOrderQuery,
    lambda: GetOrderQueryHandler(read_repo),
)
6
Use HandlerRegistry(allow_override=True) in tests so you can swap handlers without raising DuplicateHandlerError.
7
Build a MiddlewarePipeline
8
Middlewares are applied in the order they are added. The first middleware added is the outermost wrapper (executed first on the way in, last on the way out).
9
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.infrastructure.cqrs.middlewares import TransactionMiddleware, LoggingMiddleware

pipeline = (
    MiddlewarePipeline()
    .add(LoggingMiddleware())
    .add(TransactionMiddleware())
)
10
Instantiate the buses
11
from hexcore.application.cqrs.in_memory_buses import (
    InMemoryCommandBus,
    InMemoryQueryBus,
    InMemoryEventBus,
)

command_bus = InMemoryCommandBus(registry=registry, pipeline=pipeline)
query_bus   = InMemoryQueryBus(registry=registry)
event_bus   = InMemoryEventBus()
12
Configure via CQRSConfig (declarative approach)
13
For production use, define your buses and middlewares declaratively through CQRSConfig and build them with CQRSFactory. This approach lets you swap backends (e.g. Redis, RabbitMQ) without touching application code.
14
from hexcore.application.cqrs.config import CQRSConfig, BusConfig
from hexcore.application.cqrs.factory import CQRSFactory

cqrs_config = CQRSConfig(
    command_bus=BusConfig(
        middlewares=[
            "hexcore.infrastructure.cqrs.middlewares.LoggingMiddleware",
            "hexcore.infrastructure.cqrs.middlewares.TransactionMiddleware",
        ],
    ),
    query_bus=BusConfig(),   # in-memory, no middlewares
    event_bus=BusConfig(),   # in-memory, no middlewares
)

factory = CQRSFactory(config=cqrs_config, registry=registry)

command_bus = factory.create_command_bus()
query_bus   = factory.create_query_bus()
event_bus   = factory.create_event_bus()

​
CQRSConfig and BusConfig reference

​
enabled
bool
default:"True"
Master switch. When False the CQRS layer is skipped entirely.
​
command_bus
BusConfig
Configuration for the CommandBus. Defaults to an in-memory bus with TransactionMiddleware.
​
query_bus
BusConfig
Configuration for the QueryBus. Defaults to a bare in-memory bus (no middlewares).
​
event_bus
BusConfig
Configuration for the EventBus. Defaults to a bare in-memory bus.
​
serializer
str | None
default:"None"
Dotted import path to a custom ISerializer implementation. Defaults to PydanticSerializer when None.

​
BusConfig fields

​
backend
str | None
default:"None"
Dotted import path to the bus class to instantiate (e.g. "hexcore.infrastructure.cqrs.redis_bus.RedisEventBus"). When None, the default in-memory implementation is used.
​
middlewares
list[str]
default:"[]"
Ordered list of dotted import paths to AbstractMiddleware subclasses. Instantiated with no arguments, so each middleware must have a no-arg constructor.
​
options
dict[str, Any]
default:"{}"
Arbitrary keyword arguments forwarded to the bus constructor when a custom backend is specified.

​
Migration: wrapping legacy UseCase classes

If you have existing UseCase objects and want to adopt CQRS incrementally, use UseCaseCommandHandler — it adapts any UseCase into an AbstractCommandHandler without requiring a rewrite. 
from hexcore.application.cqrs.adapters import UseCaseCommandHandler
from myapp.use_cases import CreateOrderUseCase
from myapp.commands import CreateOrderCommand

# Existing use case — untouched
create_order_uc = CreateOrderUseCase(order_repository)

# Wrap it as a CQRS handler
handler = UseCaseCommandHandler(create_order_uc)

registry.register_command_handler(CreateOrderCommand, handler)
UseCaseCommandHandler.handle(command) simply calls use_case.execute(command), so the use case receives the full command object as its input argument.

​
FastAPI dependency injection

Expose the buses as FastAPI dependencies so they can be injected into routers cleanly. 
# app/dependencies.py
from functools import lru_cache
from hexcore.application.cqrs.in_memory_buses import (
    InMemoryCommandBus,
    InMemoryQueryBus,
    InMemoryEventBus,
)
from app.registry import build_registry  # your factory function
from app.config import cqrs_config

@lru_cache
def get_command_bus() -> InMemoryCommandBus:
    from hexcore.application.cqrs.factory import CQRSFactory
    registry = build_registry()
    return CQRSFactory(cqrs_config, registry).create_command_bus()

@lru_cache
def get_query_bus() -> InMemoryQueryBus:
    from hexcore.application.cqrs.factory import CQRSFactory
    registry = build_registry()
    return CQRSFactory(cqrs_config, registry).create_query_bus()

# app/routers/orders.py
from fastapi import APIRouter, Depends
from hexcore.domain.cqrs.buses import ICommandBus, IQueryBus
from app.dependencies import get_command_bus, get_query_bus
from app.commands import CreateOrderCommand
from app.queries import GetOrderQuery

router = APIRouter(prefix="/orders")

@router.post("/")
async def create_order(
    payload: CreateOrderCommand,
    bus: ICommandBus = Depends(get_command_bus),
):
    result = await bus.dispatch(payload)
    return {"order_id": str(result)}

@router.get("/{order_id}")
async def get_order(
    order_id: str,
    bus: IQueryBus = Depends(get_query_bus),
):
    return await bus.ask(GetOrderQuery(order_id=order_id))
lru_cache makes the buses singletons for the lifetime of the process. In tests, call get_command_bus.cache_clear() between test cases if you need to rebuild with a fresh registry.

Build docs developers (and LLMs) love

Get started for freeTalk to us