In CQRS everything that changes data goes through commands. Everything that reads goes through queries. The boundary is strict and never mixed. This article is about the writing half: what a command is, what a handler looks like, and where to place the checks.
Why separate reads and writes at all
Picture an ordinary service: one method get_order() reads data, another confirm_order() changes it. While the load is light, everything is fine. But as the system grows, reads and writes start getting in each other's way. Read requests want denormalized data and caching; write requests want strict transactions and business-rule checks.
CQRS solves this through separation: commands (Commands) change state through the aggregate and the UnitOfWork; queries (Queries) read through separate read models. Each path is optimized for its own task.
What a command is
A command is an intention to change state. Not a request for information, but a request to do something: "confirm the order", "create the product", "cancel the payment".
In Python commands are made with @dataclass(frozen=True):
# core/order/command/confirm_order.py
from dataclasses import dataclass
from core.order.domain.value_objects import OrderId
@dataclass(frozen=True)
class ConfirmOrder:
order_id: OrderId
idempotency_key: str
frozen=True makes the object immutable: once created, you cannot change its fields. This matters — a command describes a specific intention that must not change during execution.
A few rules for commands:
- No logic inside. A command is data, not behavior. Mapping from the incoming API request happens in the router, not here.
__post_init__is allowed only for coercing simple types into value objects.idempotency_keyis a standard field for operations that must not accidentally be performed twice. The router takes it from theIdempotency-Keyheader.
Handler structure
A command handler performs a single action in four steps: it loads the aggregate, calls a domain method, saves it, and commits the transaction.
# core/order/handler/confirm_order_handler.py
from core.order.command.confirm_order import ConfirmOrder
from core.order.domain.port import OrderRepository
from core.order.domain.value_objects import OrderId
from core.shared.uow import UnitOfWork
from core.shared.clock import Clock
class ConfirmOrderHandler:
def __init__(
self,
orders: OrderRepository,
uow: UnitOfWork,
clock: Clock,
) -> None:
self._orders = orders
self._uow = uow
self._clock = clock
async def handle(self, cmd: ConfirmOrder) -> OrderId:
async with self._uow:
# 1. Load the aggregate
order = await self._orders.by_id(cmd.order_id)
if order is None:
raise OrderNotFound(cmd.order_id)
# 2. Call the domain method — it checks the business rules
order.confirm(self._clock)
# 3. Save the aggregate
await self._orders.save(order)
# 4. Commit the transaction
await self._uow.commit()
return order.id
A few details that matter:
async with self._uow — the context manager opens a SQLAlchemy session and automatically rolls back on an exception. commit() is called explicitly in the handler, not somewhere inside the repository.
The domain method checks invariants. order.confirm() will itself raise OrderAlreadyConfirmed if the order is already confirmed. The handler need not duplicate that check.
Events are registered inside the aggregate. The order.confirm() method internally calls self._events.append(OrderConfirmed(...)). On save, the repository reads the accumulated events and writes them to the outbox for later publication.
One command — one aggregate
An important constraint: one command changes exactly one aggregate. If the business logic requires changing two, that is a signal something is off.
The bad option — changing two aggregates in one transaction:
# Don't do this — two aggregates in one UoW
async def handle(self, cmd: CreateOrder) -> OrderId:
async with self._uow:
customer = await self._customers.by_id(cmd.customer_id)
customer.increment_order_count()
await self._customers.save(customer)
order = Order.create(cmd.customer_id, cmd.items)
await self._orders.save(order)
await self._uow.commit()
return order.id
The problem: on a concurrent change to the Customer from another handler, SQLAlchemy will raise StaleDataError. Customer and Order live differently — they have different change frequencies and different reasons to change.
The correct option — one aggregate, and the event travels onward:
# One aggregate, changes propagate to others via events
async def handle(self, cmd: CreateOrder) -> OrderId:
async with self._uow:
order = Order.create(cmd.customer_id, cmd.items)
await self._orders.save(order)
await self._uow.commit()
# OrderCreated is registered inside the aggregate;
# the event handler will update Customer asynchronously
return order.id
If two objects must always change together, perhaps they are a single aggregate. If they change independently, they use an event chain or a saga.
What the handler returns
A command handler returns the minimum: the identifier of the created or changed entity, a status, or nothing (None).
# Returns the id of the created entity — fine
class CreateProductHandler:
async def handle(self, cmd: CreateProduct) -> ProductId:
async with self._uow:
product = Product.create(cmd.name, cmd.price, cmd.sku)
await self._products.save(product)
await self._uow.commit()
return product.id
# Returns None for simple operations — also fine
class CancelOrderHandler:
async def handle(self, cmd: CancelOrder) -> None: ...
A full read-DTO (for example OrderSummarySchema with all the order fields) from a command handler is a typical mistake. A command handler is responsible only for writing; reading data is the job of a separate query (query handler).
After the command, the router returns 201 Created with Location: /orders/{id}. If the client needs the full view, it makes a separate GET request:
# core/order/router.py
@router.post("/orders", status_code=201)
async def create_order(
body: CreateOrderBody,
handler: CreateOrderHandler = Depends(),
) -> CreatedOrderResponse:
order_id = await handler.handle(
CreateOrder(customer_id=body.customer_id, items=body.items)
)
return CreatedOrderResponse(id=order_id)
Where to validate
Checks split into two levels, and they must not be mixed.
The first level — the shape of the incoming request. A Pydantic schema on the router checks required fields, types, and string lengths. An error here is 422 Unprocessable Entity.
class ConfirmOrderBody(BaseModel):
idempotency_key: str = Field(min_length=1, max_length=64)
@router.post("/orders/{order_id}/confirm")
async def confirm_order(
order_id: UUID,
body: ConfirmOrderBody,
handler: ConfirmOrderHandler = Depends(),
) -> None:
await handler.handle(
ConfirmOrder(
order_id=OrderId(order_id),
idempotency_key=body.idempotency_key,
)
)
The second level — business rules. The aggregate method checks whether the operation can be performed in the current state. An error here is a domain exception, which turns into 409 Conflict or 400 Bad Request.
class Order:
def confirm(self, clock: Clock) -> None:
if self.status != OrderStatus.NEW:
raise OrderAlreadyConfirmed(self.id, self.status)
if not self.items:
raise EmptyOrderCannotBeConfirmed(self.id)
self.status = OrderStatus.CONFIRMED
self._register_event(OrderConfirmed(self.id, clock.now()))
These two layers are independent. Pydantic knows nothing about the aggregate's business state; the aggregate knows nothing about HTTP.
Common mistakes
A separate SELECT to make a decision. Sometimes people write it like this: "first I'll check whether a payment exists, then load the order". That's a problem: between the two queries the data can change, and read logic leaks into the write path. If the aggregate needs to know about the payment, payment_status should be a field of the aggregate, and order.confirm() checks it itself.
Two aggregates in one transaction — covered above. Use a saga or an event chain.
commit() inside the repository. The repository only handles saving the aggregate. The transaction is committed by the handler via UnitOfWork.
Checking an invariant in the handler instead of the aggregate. if order.status != Status.NEW: raise in the handler is logic that belongs to the aggregate. If this rule changes, it will have to be found and changed in every handler.
In short
- A command is a
@dataclass(frozen=True)with no logic inside. It describes the intention to change state. - A handler follows four steps: load the aggregate → call the domain method → save → commit the transaction.
commit()is called in the handler, not in the repository.- One command changes one aggregate. Two aggregates means a saga or wrongly sliced boundaries.
- A handler returns an
idorNone, but not a full read-DTO. - Pydantic checks the input shape, the aggregate checks the business rules. These layers are independent.
- Domain events are registered inside the aggregate, not outside.
What to read next
- Query side — the read handler through a ViewRepository and a read-only session.
- Sync via events — how an event from a command reaches the read model through the outbox.
- Read-model — the denormalized projection and how to rebuild it.
- When CQRS is justified — when it's worth applying and when plain CRUD is enough.