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.

Commands represent an explicit intent to change the state of the system. In HexCore’s CQRS layer, every state-mutating operation starts as an immutable Command object that is dispatched through InMemoryCommandBus, routed through the middleware pipeline, and finally delegated to a single registered AbstractCommandHandler. This page covers how to define commands, write handlers, wire everything together in HandlerRegistry, and dispatch safely.

​
The Command base class

Every command in HexCore inherits from hexcore.domain.cqrs.commands.Command. It is a frozen Pydantic model, meaning instances are immutable after construction and are safe to pass across layers without risk of accidental mutation. 
# hexcore/domain/cqrs/commands.py (simplified)
from uuid import UUID, uuid4
from datetime import datetime, UTC
from pydantic import BaseModel, Field, ConfigDict

class Command(BaseModel):
    command_id: UUID = Field(default_factory=uuid4)
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))

    model_config = ConfigDict(
        frozen=True,
        from_attributes=True,
    )

​
Built-in fields

​
command_id
UUID
Universally unique identifier for this command instance. Auto-generated via uuid4() — you rarely need to set this manually.
​
timestamp
datetime
UTC datetime of when the command was created. Auto-generated; useful for auditing and idempotency checks.

​
Defining a command

Add only the fields needed to express the intent. Because Command is a Pydantic model, you get automatic validation for free. 
# app/commands.py
from hexcore.domain.cqrs.commands import Command

class CreateOrderCommand(Command):
    customer_id: str
    product_ids: list[str]
    total_amount: float
Keep commands as thin as possible. They should carry only the data needed to describe the intent — not perform any logic. Business rules live in the handler.

​
AbstractCommandHandler

Every command type is handled by exactly one handler that inherits from AbstractCommandHandler[TCommand, TCommandResult]. 
# hexcore/domain/cqrs/handlers.py (simplified)
import abc
import typing as t
from .commands import Command

TCommand = t.TypeVar("TCommand", bound=Command)
TCommandResult = t.TypeVar("TCommandResult")

class AbstractCommandHandler(abc.ABC, t.Generic[TCommand, TCommandResult]):
    @abc.abstractmethod
    async def handle(self, command: TCommand) -> TCommandResult:
        raise NotImplementedError
​
TCommand
TypeVar
The concrete Command subclass this handler processes. The bus enforces a 1-to-1 mapping at dispatch time.
​
TCommandResult
TypeVar
The return type. Use None for fire-and-forget commands that produce no meaningful return value.

​
Writing a handler

# app/handlers.py
from hexcore.domain.cqrs.handlers import AbstractCommandHandler
from app.commands import CreateOrderCommand
from app.domain.order import Order
from app.repositories import IOrderRepository

class CreateOrderHandler(AbstractCommandHandler[CreateOrderCommand, str]):
    def __init__(self, order_repo: IOrderRepository) -> None:
        self._repo = order_repo

    async def handle(self, command: CreateOrderCommand) -> str:
        order = Order.create(
            customer_id=command.customer_id,
            product_ids=command.product_ids,
            total_amount=command.total_amount,
        )
        await self._repo.save(order)
        return str(order.id)

​
Registering and dispatching

1
Register the handler in HandlerRegistry
2
from hexcore.application.cqrs.registry import HandlerRegistry
from app.commands import CreateOrderCommand
from app.handlers import CreateOrderHandler
from app.repositories import SqlOrderRepository

registry = HandlerRegistry()
registry.register_command_handler(
    CreateOrderCommand,
    CreateOrderHandler(SqlOrderRepository(session)),
)
3
The registry returns self from register_command_handler, so you can chain registrations fluently:
4
registry = (
    HandlerRegistry()
    .register_command_handler(CreateOrderCommand, CreateOrderHandler(order_repo))
    .register_command_handler(CancelOrderCommand, CancelOrderHandler(order_repo))
)
5
Build the bus
6
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.infrastructure.cqrs.middlewares import TransactionMiddleware

pipeline = MiddlewarePipeline().add(TransactionMiddleware())
command_bus = InMemoryCommandBus(registry=registry, pipeline=pipeline)
7
Dispatch the command
8
command = CreateOrderCommand(
    customer_id="cust-42",
    product_ids=["prod-1", "prod-2"],
    total_amount=99.90,
)

order_id = await command_bus.dispatch(command)
print(f"Order created: {order_id}")

​
Complete example

# ── Domain ──────────────────────────────────────────────────────
from hexcore.domain.cqrs.commands import Command
from hexcore.domain.cqrs.handlers import AbstractCommandHandler

class PlaceOrderCommand(Command):
    customer_id: str
    items: list[str]

class PlaceOrderHandler(AbstractCommandHandler[PlaceOrderCommand, str]):
    async def handle(self, command: PlaceOrderCommand) -> str:
        # Business logic here
        order_id = f"order-{command.command_id}"
        print(f"Placing order for {command.customer_id}: {command.items}")
        return order_id

# ── Wiring ───────────────────────────────────────────────────────
from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.infrastructure.cqrs.middlewares import LoggingMiddleware, TransactionMiddleware

registry = HandlerRegistry()
registry.register_command_handler(PlaceOrderCommand, PlaceOrderHandler())

pipeline = MiddlewarePipeline().add(LoggingMiddleware()).add(TransactionMiddleware())
bus = InMemoryCommandBus(registry=registry, pipeline=pipeline)

# ── Dispatch ─────────────────────────────────────────────────────
import asyncio

async def main():
    cmd = PlaceOrderCommand(customer_id="cust-1", items=["apple", "banana"])
    result = await bus.dispatch(cmd)
    print(f"Result: {result}")

asyncio.run(main())

​
TransactionMiddleware and commands

CQRSConfig enables TransactionMiddleware on the command bus by default: 
# hexcore/application/cqrs/config.py
class CQRSConfig(BaseModel):
    command_bus: BusConfig = Field(
        default_factory=lambda: BusConfig(
            middlewares=[
                "hexcore.infrastructure.cqrs.middlewares.TransactionMiddleware"
            ]
        )
    )
This means every command dispatched through a factory-built bus is automatically wrapped in a Unit of Work transaction. The commit() is called after the handler returns successfully; on any exception the UoW rolls back and re-raises. You do not need to manage transactions manually inside handlers.
TransactionMiddleware uses uow_factory to obtain a UoW context manager. If you inject a custom factory, it takes precedence over the default SQL/NoSQL auto-detection heuristic.

​
Error handling

HandlerNotFoundError

Raised by the registry when dispatch() is called for a command type with no registered handler. Always register all command types before starting the application.

DuplicateHandlerError

Raised when you try to register a second handler for the same command type. Pass allow_override=True to the registry constructor to suppress this in tests.

Build docs developers (and LLMs) love

Get started for freeTalk to us