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.
Command Query Responsibility Segregation (CQRS) is an architectural pattern that separates the models used for writing (commands) from those used for reading (queries). HexCore v2 implements CQRS natively with three independent buses, a central handler registry, and a middleware pipeline — giving you clear boundaries between side-effecting operations and pure reads, all wired with zero framework lock-in. This page covers every bus port, explains why queries are treated differently from commands, and shows you how to register handlers and dispatch messages in real code.
The three buses
HexCore defines three abstract bus ports in hexcore/domain/cqrs/buses.py. Each is an independent abc.ABC so that you can substitute any implementation — in-memory, Redis, RabbitMQ, or your own — without changing application-layer code.
AbstractCommandBus — write side
A command expresses an intent to change state. The bus routes it to exactly one handler, executes any middleware, and returns the handler’s result.
# hexcore/domain/cqrs/buses.py
class AbstractCommandBus(abc.ABC):
@abstractmethod
async def dispatch(self, command: "Command") -> Any:
"""
Routes a command to its registered handler.
Raises HandlerNotFoundError if no handler is registered.
"""
raise NotImplementedError
AbstractQueryBus — read side
A query expresses a request for information with no side effects. The bus routes it to exactly one handler, which returns a typed result. Queries never use background execution — reads are always synchronous relative to the caller.
# hexcore/domain/cqrs/buses.py
class AbstractQueryBus(abc.ABC):
@abstractmethod
async def ask(self, query: "Query[TResult]") -> TResult:
"""
Routes a query to its registered handler and returns the result.
Raises HandlerNotFoundError if no handler is registered.
"""
raise NotImplementedError
AbstractEventBus — domain events
An event bus distributes domain events to zero or more subscribed handlers. Unlike command and query buses, the event bus supports fan-out: multiple parts of the system can react to the same event independently.
# hexcore/domain/cqrs/buses.py
class AbstractEventBus(abc.ABC):
@abstractmethod
async def publish(self, event: DomainEvent) -> None:
"""Publishes an event to all subscribed handlers."""
raise NotImplementedError
@abstractmethod
def subscribe(
self,
event_type: type[DomainEvent],
handler: Callable[[DomainEvent], Awaitable[None]],
) -> None:
"""Registers a handler for the given event type."""
raise NotImplementedError
Commands in depth
Defining a command
Command is a frozen Pydantic model. Subclass it and add the fields your use-case requires. Every command automatically gets a command_id (UUID) and timestamp (UTC datetime).
from hexcore.domain.cqrs.commands import Command
class CreateOrderCommand(Command):
customer_id: UUID
product_ids: list[UUID]
shipping_address: str
Writing a command handler
AbstractCommandHandler[TCommand, TResult] declares a single abstract handle method. Return None for fire-and-forget commands.
from hexcore.domain.cqrs.handlers import AbstractCommandHandler
from hexcore.domain.uow import IUnitOfWork
class CreateOrderHandler(AbstractCommandHandler[CreateOrderCommand, OrderDTO]):
def __init__(self, uow: IUnitOfWork) -> None:
self.uow = uow
async def handle(self, command: CreateOrderCommand) -> OrderDTO:
async with self.uow:
repo = self.uow.repositories["orders"]
order = Order(
customer_id=command.customer_id,
product_ids=command.product_ids,
shipping_address=command.shipping_address,
)
order.register_event(OrderCreatedEvent(entity_id=order.id, entity_data=order))
await repo.save(order)
await self.uow.commit()
return OrderDTO.model_validate(order)
Queries in depth
Defining a query
Query[TResult] is a frozen Pydantic model. The generic type parameter documents the return type so that the query bus and type checkers can infer the result.
from hexcore.domain.cqrs.queries import Query
class GetOrderByIdQuery(Query[OrderDTO]):
order_id: UUID
class ListOrdersQuery(Query[list[OrderDTO]]):
customer_id: UUID
limit: int = 20
offset: int = 0
Writing a query handler
from hexcore.domain.cqrs.handlers import AbstractQueryHandler
class GetOrderByIdHandler(AbstractQueryHandler[GetOrderByIdQuery, OrderDTO]):
def __init__(self, repository: IBaseRepository[Order]) -> None:
self.repository = repository
async def handle(self, query: GetOrderByIdQuery) -> OrderDTO:
order = await self.repository.get_active_by_id(query.order_id)
return OrderDTO.model_validate(order)
Queries must never mutate state. Do not call repo.save(), repo.delete(), or uow.commit() inside a query handler. This is the single rule that makes CQRS work.
Registering handlers
HandlerRegistry maps command and query types to their handlers. It supports both eager (direct instance) and lazy (factory function) registration:
from hexcore.application.cqrs.registry import HandlerRegistry
registry = HandlerRegistry()
# Eager registration — handler is instantiated immediately
registry.register_command_handler(CreateOrderCommand, CreateOrderHandler(uow=uow))
registry.register_query_handler(GetOrderByIdQuery, GetOrderByIdHandler(repository=repo))
# Lazy registration — factory is called on first dispatch (useful with DI containers)
registry.register_command_handler(
CreateOrderCommand,
lambda: container.get(CreateOrderHandler),
)
register_command_handler returns self:
registry = (
HandlerRegistry()
.register_command_handler(CreateOrderCommand, CreateOrderHandler(uow))
.register_command_handler(CancelOrderCommand, CancelOrderHandler(uow))
.register_query_handler(GetOrderByIdQuery, GetOrderByIdHandler(repo))
)
In-memory buses (default implementation)
For development, testing, and simple deployments, HexCore ships three in-memory bus implementations in hexcore.application.cqrs.in_memory_buses.
InMemoryCommandBus
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
from hexcore.application.cqrs.registry import HandlerRegistry
registry = HandlerRegistry()
registry.register_command_handler(CreateOrderCommand, CreateOrderHandler(uow))
bus = InMemoryCommandBus(registry=registry)
# Dispatch a command and receive its result
result: OrderDTO = await bus.dispatch(CreateOrderCommand(
customer_id=customer.id,
product_ids=[product.id],
shipping_address="123 Main St",
))
InMemoryQueryBus
from hexcore.application.cqrs.in_memory_buses import InMemoryQueryBus
query_bus = InMemoryQueryBus(registry=registry)
order: OrderDTO = await query_bus.ask(GetOrderByIdQuery(order_id=result.id))
InMemoryEventBus
from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
from hexcore.domain.events import DomainEvent
event_bus = InMemoryEventBus()
# Subscribe a handler
async def send_confirmation_email(event: OrderCreatedEvent) -> None:
print(f"Sending email for order {event.entity_id}")
event_bus.subscribe(OrderCreatedEvent, send_confirmation_email)
# Publish (all subscribers are called in sequence)
await event_bus.publish(OrderCreatedEvent(entity_id=order_id, entity_data=order))
Message flow diagram
Below is a narrative of the full CQRS flow in a FastAPI endpoint:
HTTP POST /orders
│
▼
FastAPI router
creates CreateOrderCommand
│
▼
InMemoryCommandBus.dispatch(cmd)
│
├─ MiddlewarePipeline runs (e.g. TransactionMiddleware)
│
▼
CreateOrderHandler.handle(cmd)
│
├─ Constructs Order entity
├─ Calls repo.save(order)
├─ Registers OrderCreatedEvent on entity
▼
uow.commit()
│
├─ DB transaction committed
├─ Calls pull_domain_events() on all tracked entities
▼
EventBus.publish(OrderCreatedEvent)
│
├─ send_confirmation_email(event) ← synchronous subscriber
└─ project_order_to_mongo(event) ← synchronous subscriber
(or background if @background_handler)
HTTP GET /orders/{id}
│
▼
FastAPI router
creates GetOrderByIdQuery
│
▼
InMemoryQueryBus.ask(query)
│
▼
GetOrderByIdHandler.handle(query)
│
▼
Returns OrderDTO directly (no UoW, no events)
Smart background routing
HexCore’s InMemoryCommandBus supports smart routing: if a command class is decorated with @background_command, the bus automatically enqueues it to a task queue instead of executing it locally. Your application code does not need to change.
from hexcore.domain.cqrs.decorators import background_command, background_handler
# This command will always execute in the background worker
@background_command(queue="high_priority")
class GenerateReportCommand(Command):
user_id: UUID
report_type: str
# This event handler will run in the background for every published event
@background_handler(queue="notifications")
async def send_email_on_order_created(event: OrderCreatedEvent) -> None:
# Called by Celery/Procrastinate worker, not the API process
await email_client.send(...)
CeleryEnqueuer (or ProcrastinateEnqueuer) to the bus:
from hexcore.infrastructure.task_queues.celery_adapter import CeleryEnqueuer
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
from celery import Celery
app = Celery("myapp", broker="redis://localhost:6379/0")
enqueuer = CeleryEnqueuer(app)
serializer = PydanticSerializer()
command_bus = InMemoryCommandBus(
registry=registry,
enqueuer=enqueuer,
serializer=serializer,
)
The InMemoryQueryBus intentionally does not accept an enqueuer. Queries are always executed synchronously because the caller needs the result before it can continue. Background execution would break the read-response contract.
Error handling
Both buses raise HandlerNotFoundError from hexcore.domain.cqrs.exceptions if no handler is registered for a given command or query type. Catch it at the outermost boundary (e.g. an API exception handler) to return a meaningful HTTP 500:
from hexcore.domain.cqrs.exceptions import HandlerNotFoundError
try:
result = await bus.dispatch(some_command)
except HandlerNotFoundError as exc:
logger.error("No handler registered: %s", exc)
raise HTTPException(status_code=500, detail="Internal configuration error")