В CQRS всё, что меняет данные, идёт через команды. Всё, что читает — через запросы. Граница жёсткая и не смешивается. Эта статья — про пишущую половину: что такое команда, как выглядит обработчик и где расставить проверки.
Зачем вообще разделять чтение и запись
Представь обычный сервис: один метод get_order() читает данные, другой confirm_order() их меняет. Пока нагрузка небольшая — всё нормально. Но с ростом системы чтение и запись начинают мешать друг другу. Запросы на чтение хотят денормализованных данных и кэша, запросы на запись — строгих транзакций и проверки бизнес-правил.
CQRS решает это разделением: команды (Commands) меняют состояние через агрегат и UnitOfWork; запросы (Queries) читают через отдельные read-модели. Каждый путь оптимизирован под свою задачу.
Что такое команда
Команда — это намерение изменить состояние. Не запрос информации, а просьба что-то сделать: «подтверди заказ», «создай продукт», «отмени платёж».
В Python команды делают через @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 делает объект неизменяемым: после создания поменять поля нельзя. Это важно — команда описывает конкретное намерение, которое не должно меняться в процессе выполнения.
Несколько правил для команд:
- Никакой логики внутри. Команда — данные, не поведение. Маппинг из входящего запроса API происходит в роутере, а не здесь.
__post_init__допустим только для приведения простых типов к value object-ам.idempotency_key— стандартное поле для операций, которые нельзя случайно выполнить дважды. Роутер берёт его из заголовкаIdempotency-Key.
Структура обработчика
Обработчик команды (handler) выполняет одно действие в четыре шага: загружает агрегат, вызывает доменный метод, сохраняет, фиксирует транзакцию.
Четыре шага и ни одного бизнес-правила в самом обработчике: правила проверяет агрегат, события в outbox пишет репозиторий.
# 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. Загрузить агрегат
order = await self._orders.by_id(cmd.order_id)
if order is None:
raise OrderNotFound(cmd.order_id)
# 2. Вызвать доменный метод — он проверяет бизнес-правила
order.confirm(self._clock)
# 3. Сохранить агрегат
await self._orders.save(order)
# 4. Зафиксировать транзакцию
await self._uow.commit()
return order.id
Несколько деталей, которые важны:
async with self._uow — контекстный менеджер открывает сессию SQLAlchemy и при исключении автоматически делает откат. commit() вызывается явно в обработчике, не где-то внутри репозитория.
Доменный метод проверяет инварианты. order.confirm() сам бросит OrderAlreadyConfirmed, если заказ уже подтверждён. Обработчику не нужно дублировать эту проверку.
События регистрируются внутри агрегата. Метод order.confirm() вызывает внутри себя self._events.append(OrderConfirmed(...)). Репозиторий при save читает накопленные события и записывает их в outbox для последующей публикации.
Одна команда — один агрегат
Важное ограничение: одна команда меняет ровно один агрегат. Если бизнес-логика требует изменить два — это сигнал, что что-то не так.
Плохой вариант — менять два агрегата в одной транзакции:
# Так делать не нужно — два агрегата в одном 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
Проблема: при конкурентном изменении Customer из другого обработчика SQLAlchemy выбросит StaleDataError — если у Customer включена оптимистичная версия (version_id_col); без неё второе изменение молча затрёт первое. Customer и Order живут по-разному — у них разная частота изменений и разные причины меняться.
Правильный вариант — один агрегат, событие летит дальше:
# Один агрегат, изменения в другие распространяются через события
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 зарегистрирован внутри агрегата;
# обработчик события обновит Customer асинхронно
return order.id
Если два объекта должны меняться всегда вместе, возможно, они один агрегат. Если меняются независимо — используют событийную цепочку или сагу.
Что возвращает обработчик
Обработчик команды возвращает минимум: идентификатор созданной или изменённой сущности, статус, либо ничего (None).
# Возвращает id созданной сущности — нормально
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
# Возвращает None для простых операций — тоже нормально
class CancelOrderHandler:
async def handle(self, cmd: CancelOrder) -> None: ...
Полный read-DTO (например OrderSummarySchema со всеми полями заказа) из обработчика команды — типичная ошибка. Обработчик команды отвечает только за запись; читать данные — задача отдельного запроса (query-handler).
Роутер после команды возвращает 201 Created с Location: /orders/{id}. Если клиенту нужен полный вид — он делает отдельный GET-запрос:
# 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)
Решение зависит от чужого агрегата
Правило «читать состояние для решения только из своего агрегата» оставляет открытым главный практический случай: нужное поле физически принадлежит другому агрегату, и втащить его в свой нельзя. Например, «нельзя подтвердить заказ, если у покупателя просрочена оплата предыдущего»: просрочка живёт в платежах, а не в заказе. Что делать, по возрастанию цены и надёжности.
1. Передать решение в команду, а не данные. Самый дешёвый и самый недооценённый приём: проверку делает тот, кто её умеет, до команды, и в команду приходит уже готовый ответ.
# adapters/in_/http/orders/order_router.py: до транзакции команды
eligibility = await payments.eligibility_for(customer_id) # чужой контекст
order_id = await confirm_order.handle(
ConfirmOrder(order_id=order_id, payment_allowed=eligibility.allowed, idempotency_key=key) # готовый факт
)
Агрегат при этом принимает решение сам, но по переданному факту, а не по данным чужого агрегата. Правило соблюдено, транзакция короткая, внешний вызов вне неё. Ограничение: факт может устареть между проверкой и выполнением, и здесь нужен следующий пункт.
2. Скопировать нужный факт в свой агрегат через событие. Если решение требуется часто, чужой факт переносят к себе: платежи публикуют «у покупателя просрочка», заказы держат у себя поле и обновляют его по событию. Это не дублирование данных, а проекция чужого факта в свою модель, и она стоит ровно того, чего стоит любая проекция: отставание на доли секунды и необходимость уметь пересобрать. Признак, что приём подходит: факт меняется редко, а читается часто.
3. Принять решение оптимистично и компенсировать. Когда проверить заранее нельзя (резерв на складе, лимит, баланс), команда выполняется, а чужая сторона подтверждает или отклоняет, то есть получается сага. Пользователь видит промежуточное состояние («ожидает подтверждения»), и это честнее, чем блокировать два агрегата в одной транзакции.
4. Признать, что границы неверные. Если решение всегда требует чужого состояния и промежуточное состояние недопустимо, значит, два агрегата на самом деле один. Признак: каждое изменение одного требует одновременного изменения другого.
Чего делать не стоит: читать чужую таблицу напрямую из своего обработчика, держать чужой агрегат в транзакции «только для чтения» и делать вызов в чужой сервис внутри транзакции. Как выбрать: редко меняется и часто нужно, копировать событием; нужно однократно и точно, передавать решением до команды; нельзя знать заранее, сага.
Где валидировать
Проверки делятся на два уровня, и их нельзя смешивать.
Первый уровень — форма входящего запроса. Pydantic-схема на роутере проверяет обязательные поля, типы, длины строк. Ошибка здесь — 422 Unprocessable Entity.
class ConfirmOrderBody(BaseModel):
idempotency_key: str = Field(min_length=1, max_length=64)
@router.post("/orders/{order_id}/confirm", status_code=204)
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,
)
)
Второй уровень — бизнес-правила. Метод агрегата проверяет, можно ли выполнить операцию в текущем состоянии. Ошибка здесь — доменное исключение, которое превращается в 409 Conflict или 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()))
Эти два слоя независимы. Pydantic не знает про бизнес-состояние агрегата, агрегат не знает про HTTP.
Частые ошибки
Отдельный SELECT для принятия решения. Иногда пишут так: «сначала проверю, есть ли платёж, потом загружу заказ».
# так делать не нужно
async def handle(self, cmd: ConfirmOrder) -> OrderId:
has_payment = await self._payments.exists_by_order_id(cmd.order_id) # отдельный read
if not has_payment:
raise PaymentRequired(cmd.order_id)
async with self._uow:
order = await self._orders.by_id(cmd.order_id)
order.confirm(self._clock)
await self._orders.save(order)
await self._uow.commit()
return order.id
Проблема двойная: между первым и вторым запросом данные могут измениться (гонка без блокировки), и логика чтения просочилась в обработчик записи. Если статус платежа важен для подтверждения, он должен быть полем агрегата, и order.confirm() проверит его сам.
Два агрегата в одной транзакции — разобрано выше. Используй сагу или событийную цепочку.
# частая ошибка: два агрегата под одной транзакцией
async def handle(self, cmd: TransferMoney) -> None:
async with self._uow:
source = await self._accounts.by_id_for_update(cmd.from_id)
target = await self._accounts.by_id_for_update(cmd.to_id)
source.debit(cmd.amount)
target.credit(cmd.amount)
await self._accounts.save(source)
await self._accounts.save(target)
await self._uow.commit()
# правильно: сага из двух локальных команд
# 1. DebitAccount → AccountDebited
# 2. CreditAccount по событию
# при ошибке компенсирующая команда CreditAccount (возврат)
commit() внутри репозитория. Репозиторий занимается только сохранением агрегата. Транзакцию фиксирует обработчик через UnitOfWork.
Проверка инварианта в обработчике вместо агрегата. if order.status != Status.NEW: raise в handler-е — это логика, которая принадлежит агрегату. Если это правило изменится, его нужно будет найти и поменять в каждом обработчике.
Глубже: версия агрегата: откуда берётся и что видит проигравшийрасширенное
Идемпотентность проекций в статьях про синхронизацию держится на колонке version и поле aggregate_version в событии, а command side учит только FOR UPDATE. Откуда версия берётся и кто её увеличивает, стоит показать здесь, потому что это часть обработчика.
В SQLAlchemy версию ведёт сам маппер: колонка version: Mapped[int] и __mapper_args__ = {"version_id_col": version}. При каждом UPDATE ORM добавляет к условию WHERE version = :старая и увеличивает её на единицу; если строк обновилось ноль, кто-то успел раньше, и flush бросает StaleDataError. Это оптимистичная блокировка: два обработчика читают заказ с версией 7, оба меняют, первый пишет версию 8, второй падает.
Что видит проигравший: исключение в обработчике команды, которое реестр ошибок переводит в 409 Conflict с кодом CONCURRENT_UPDATE. Клиент перечитывает агрегат и повторяет команду, если она ещё имеет смысл; повторять автоматически внутри сервиса нельзя, потому что решение могло измениться.
Та же версия уходит в событие: агрегат, регистрируя OrderConfirmed, кладёт в него aggregate_version после инкремента, и потребитель проекции по ней отбрасывает устаревшие события. Пессимистичная блокировка with_for_update() при этом не исчезает: её берут там, где конфликт частый и повтор дорог, а версию держат всегда.
Глубже: ключ идемпотентности команды: где хранится и что вернуть на повторрасширенное
Поле idempotency_key появилось в команде выше и больше не упоминалось, а для денежных команд это центральная деталь: клиент отправил «оплатить», ответ потерялся, клиент повторил, и второй платёж это инцидент.
Ключ генерирует клиент, один UUID на попытку операции, и присылает заголовком Idempotency-Key; роутер кладёт его в команду. Обработчик первым делом вставляет строку в таблицу command_key(key, actor_id, request_hash, response, created_at) той же транзакцией, что и работа. На повторе вставка упирается в уникальный индекс, и обработчик отвечает сохранённым ответом с тем же статусом, не выполняя команду. Если ключ тот же, а тело другое (request_hash не сошёлся), это ошибка клиента, 422 с кодом IDEMPOTENCY_KEY_REUSED. Если первая попытка ещё выполняется, вторая получает 409 и повторяет позже.
Хранят ответ, а не только факт: клиенту нужен тот же order_id, что в первый раз. Ключи живут ограниченный срок и чистятся пачками; сроки и уборку разбирает статья про идемпотентность при остановке.
Глубже: границы транзакции в обработчикерасширенное
async with self._uow стоит в каждом примере, и стоит сказать, что он делает в четырёх ситуациях, где интуиция обманывает.
Внешний вызов внутри транзакции. Обработчик в транзакции зовёт платёжный сервис по HTTP на три секунды: всё это время соединение из пула занято и строки заблокированы FOR UPDATE. Под нагрузкой пул заканчивается первым. Внешние вызовы делают до транзакции (прочитать курс, проверить лимит) или после неё через outbox, но не внутри.
Второй begin(). SQLAlchemy открывает транзакцию сама при первом запросе, и async with session.begin() после уже выполненного SELECT падает с «transaction is already begun». UnitOfWork должен начинать транзакцию на свежей сессии, одна сессия на команду.
Доступ к полям после коммита. С настройкой expire_on_commit=True первое обращение к order.id после commit() идёт в базу, а в async-коде это MissingGreenlet; фабрику сессий создают с expire_on_commit=False, об этом статья про SQLAlchemy во FastAPI.
События после коммита. Публиковать OrderConfirmed в Kafka внутри транзакции нельзя: откат не отзовёт сообщение. Репозиторий пишет события в outbox той же транзакцией, а relay отправляет после фиксации; так гарантия «событие есть тогда и только тогда, когда есть изменение» держится без распределённых транзакций.
Коротко
- Команда —
@dataclass(frozen=True)без логики внутри. Описывает намерение изменить состояние. - Обработчик следует четырём шагам: загрузить агрегат → вызвать доменный метод → сохранить → зафиксировать транзакцию.
commit()вызывается в обработчике, не в репозитории.- Одна команда меняет один агрегат. Два агрегата — это сага или неверно нарезанные границы.
- Обработчик возвращает
idилиNone, но не полный read-DTO. - Pydantic проверяет форму входа, агрегат проверяет бизнес-правила. Эти слои независимы.
- Доменные события регистрируются внутри агрегата, не снаружи.
- Версию агрегата ведёт
version_id_col, проигравший получаетStaleDataErrorи409; ключ идемпотентности команды хранят с ответом в той же транзакции; внутри транзакции нет внешних вызовов и второгоbegin(), события уходят через outbox.
Что почитать дальше
- Query side — read-handler через ViewRepository и read-only сессию.
- Sync через события — как событие из команды доходит до read-модели через outbox.
- Read-model — денормализованная проекция и её восстановление.
- Когда CQRS оправдан — когда стоит применять, а когда достаточно простого CRUD.