CQRS и event sourcing
Зачем разделять модели для записи и для чтения (CQRS) и когда это лишнее; как хранить не состояние, а события (event sourcing): свёртка, хранилище событий, проекции, снимки и версионирование событий; что это даёт и чего стоит. Учебная реализация потока из событий и проекции «расписание студента» на Python.
Зачем это нужно
Заголовок раздела «Зачем это нужно»Методист онлайн-школы пишет в чат разработчикам: «Борис говорит, что записывался на поток по Python и даже начал
оплачивать, а теперь записи нет и место занято. Что случилось?» В базе у Бориса статус CANCELLED. Когда он
появился? Кто отменил запись — Борис, поддержка или автоматика через 48 часов? Кто занял место? Таблица хранит только
то, что есть сейчас, и на вопрос «как мы сюда попали» не отвечает. Мы уже видели эту беду в
уроке DDD.1.1: «удалить заказ» вместо «запись сгорает» — и история пропала.
Вторая история — с витриной. Страницу каталога со свободными местами по всем потокам открывают тысячи раз в день,
а записываются на потоки десятки раз. Чтобы показать 40 потоков, код загружает 40 агрегатов Cohort со всеми
записями и листами ожидания — модель, построенная, чтобы защищать правила, работает справочником. А экрану «моё
расписание» нужны потоки разных курсов одного студента, то есть данные сразу многих агрегатов.
Эти две проблемы решают две идеи, которые часто называют вместе. CQRSразделение ответственности команд и запросов (CQRS)EN Command Query Responsibility SegregationИдея, которую около 2010 года популяризировал Грег Янг: для изменения данных и для их чтения система держит разные модели. Запись идёт через агрегаты с инвариантами, чтение — через простые модели, удобные для экранов. Не требует ни отдельной базы, ни event sourcing. разделяет модель для изменения данных и модель для их чтения. Event sourcingevent sourcing (состояние как журнал событий)EN event sourcingСпособ хранения: вместо текущего состояния агрегата сохраняют все его события по порядку, а состояние получают их свёрткой. Журнал только дописывают — прошлое не меняется. Даёт аудит и историю, но усложняет систему. Встречаются переводы «порождение событий» и «событийное хранение». хранит не состояние, а события, из которых оно получилось. Друг без друга они тоже работают. В этом уроке разберём обе идеи, честно посчитаем их цену и соберём учебную реализацию на Python: поток, восстановленный из событий, и две проекции. В браузерном задании вы напишете свёртку и проекцию «свободные места» сами.
Команды и запросы
Заголовок раздела «Команды и запросы»CQS: метод либо меняет, либо отвечает
Заголовок раздела «CQS: метод либо меняет, либо отвечает»Бертран Мейер, автор языка Eiffel, сформулировал принцип разделения команд и запросовразделение команд и запросов (CQS)EN Command-Query SeparationПринцип Бертрана Мейера для отдельных методов и классов: метод либо меняет состояние (команда), либо отвечает на вопрос (запрос), но не то и другое сразу. Узнать, горит ли свет, можно, не щёлкая выключателем. (Command-Query Separation, CQS): каждый метод — либо команда, которая меняет состояние и ничего не возвращает, либо запрос, который возвращает ответ и ничего не меняет. Авторы «Cosmic Python» пересказывают его так: должно быть можно спросить, горит ли свет, не щёлкая выключателем.
В модели школы cohort.enroll(student_id, now) — команда, cohort.free_seats() — запрос. Представьте, что кто-то
«оптимизировал» free_seats(): перед подсчётом он заодно отменяет неоплаченные записи старше 48 часов. Витрина
спросила, сколько мест, — и три записи тихо отменились. Отладка такого поведения стоит дней, потому что никто не
ждёт изменений от вопроса.
CQRS: две модели вместо одной
Заголовок раздела «CQRS: две модели вместо одной»Грег Янг около 2010 года перенёс идею с методов на всю систему и назвал её CQRS — Command Query Responsibility Segregation, разделение ответственности команд и запросов. В одной из статей того времени он описывал её почти буднично: там, где был один объект, сделать два — один для команд, другой для запросов.
Дело в том, что запись и чтение — разные задачи с разными требованиями:
| Запись (команды) | Чтение (запросы) | |
|---|---|---|
| Что делает | меняет состояние по правилам | показывает данные экрану или отчёту |
| Модель | агрегат Cohort с инвариантами | плоская таблица «как на экране» |
| Как часто в школе | десятки записей на поток в день | тысячи просмотров витрины |
| Согласованность | строгая: два студента не займут одно место | может отставать на секунды |
| Кэширование | почти невозможно | легко |
Авторы «Cosmic Python» показывают ту же асимметрию на примере своей работы в мебельном интернет-магазине MADE.com: в оживлённый день — около сотни заказов в час и около сотни просмотров товаров в секунду. Модель предметной области со всеми её правилами нужна первым, а не вторым.
Почему чтению можно отставать
Заголовок раздела «Почему чтению можно отставать»Самый частый страх перед CQRS: «а вдруг витрина покажет устаревшие данные?» Покажет — но так происходит и без CQRS. Витрина показала Вере «1 свободное место». Пока Вера заполняла форму, место занял Глеб. Как только страница нарисована, данные на ней уже могут быть устаревшими, и никакая архитектура этого не отменит.
Поэтому правило звучит так: чтение может отставать, запись — никогда. Когда Вера нажмёт «Записаться», агрегат
проверит вместимость по актуальным данным и откажет NoFreeSeats, а интерфейс предложит лист ожидания.
Такое временное расхождение называют согласованностью в конечном счётесогласованность в конечном счётеEN eventual consistencyДанные в разных местах (агрегат и модель чтения, два контекста) какое-то время расходятся, но без новых изменений сходятся к одному значению. Плата за независимость частей системы; бизнес обычно и так живёт с задержками и исправлениями.
(eventual consistency): модель чтения отстаёт от агрегата, но сходится к нему.
Ступени CQRS
Заголовок раздела «Ступени CQRS»CQRS не означает «две базы данных». Это шкала, и подниматься по ней стоит ровно настолько, насколько нужно.
| Ступень | Как устроено чтение | Согласованность | Цена |
|---|---|---|---|
| 0. Без CQRS | экраны читают через репозиторий и агрегаты | строгая | медленно для сложных экранов, агрегаты обрастают методами «для экрана» |
| 1. Отдельные запросы | функции-запросы читают те же таблицы напрямую, в обход агрегатов | строгая | запросы надо править при смене схемы |
| 2. Таблицы для чтения | денормализованные таблицы обновляются в той же транзакции, что и агрегат | строгая | запись чуть медленнее |
| 3. Отдельное хранилище | обработчики событий обновляют модели чтения, возможно в другой базе | в конечном счёте | больше движущихся частей, нужна перестройка |
Таблица близка к сравнению вариантов в главе 12 «Cosmic Python». Ступень 1 — уже CQRS, хотя в ней нет ни событий,
ни второй базы. Вот запрос для витрины каталога на sqlite3 из стандартной библиотеки: он не загружает ни
одного агрегата и считает места одним SQL-запросом.
import sqlite3
conn = sqlite3.connect(":memory:")conn.executescript(""" CREATE TABLE cohorts (cohort_id TEXT PRIMARY KEY, capacity INTEGER, starts_at TEXT); CREATE TABLE enrollments (cohort_id TEXT, student_id TEXT, status TEXT); INSERT INTO cohorts VALUES ('python-2026-11', 3, '2026-11-02 10:00'), ('ds-2026-12', 2, '2026-12-07 10:00'); INSERT INTO enrollments VALUES ('python-2026-11', 'anna', 'CONFIRMED'), ('python-2026-11', 'boris', 'CANCELLED'), ('ds-2026-12', 'vera', 'PENDING_PAYMENT'), ('ds-2026-12', 'gleb', 'CONFIRMED');""")
def free_seats_by_cohort(conn): """Запрос для витрины каталога: свободные места на всех потоках — один SQL, без агрегатов.""" return conn.execute(""" SELECT c.cohort_id, c.starts_at, c.capacity - COUNT(e.student_id) AS free_seats FROM cohorts AS c LEFT JOIN enrollments AS e ON e.cohort_id = c.cohort_id AND e.status != 'CANCELLED' GROUP BY c.cohort_id ORDER BY c.starts_at """).fetchall()
for cohort_id, starts_at, free in free_seats_by_cohort(conn): print(f"{cohort_id:15} старт {starts_at} свободно мест: {free}")python-2026-11 старт 2026-11-02 10:00 свободно мест: 2ds-2026-12 старт 2026-12-07 10:00 свободно мест: 0Правило «места занимают записи, кроме отменённых» здесь повторено в SQL — это цена ступени 1. Зато агрегат
Cohort остаётся маленьким и занимается только правилами, а витрина получает ровно те столбцы, что ей нужны.
Когда CQRS окупается, а когда это лишнее
Заголовок раздела «Когда CQRS окупается, а когда это лишнее»Окупается, когда чтений намного больше, чем записей, когда экран собирает данные многих агрегатов («расписание студента» — потоки разных курсов), когда для чтения нужны другие хранилища (полнотекстовый поиск, кэш) и когда модель записи сложная, а экраны простые.
Лишнее — в CRUD-частях системы (Каталог школы из урока DDD.1.3), когда экран показывает ровно один агрегат и когда отставание модели чтения будет путать пользователей. Классическая жалоба: студент записался, обновил страницу — а в расписании записи нет, проекция ещё не догнала.
Три мифа, которые стоит отбросить: CQRS не требует двух баз; CQRS не требует event sourcing; CQRS не обязан охватывать всю систему — его применяют в одном контексте и даже для одного экрана.
Event sourcing: храним события, а не состояние
Заголовок раздела «Event sourcing: храним события, а не состояние»Банк не хранит только остаток на вашем счёте. Он хранит все операции, а остаток — это их сумма. Шахматная партия записывается ходами, и из записи можно восстановить позицию на любом ходу. Event sourcing делает так же с агрегатом: вместо строки «Анна — подтверждена» хранится журнал «поток открыт, Анна записалась, Борис записался, Анна подтвердила запись, Борис отменил запись, место освободилось», а текущее состояние вычисляется из него.
Мартин Фаулер описал этот шаблон в статье «Event Sourcing» (2005): каждое изменение состояния приложения фиксируется объектом-событием, события хранятся в том порядке, в каком происходили, и журнал становится главным источником правды. Из него можно пересобрать состояние целиком, узнать состояние на любой момент в прошлом и переиграть события, если в обработке нашли ошибку. Широкую известность шаблон получил вместе с CQRS через доклады и статьи Грега Янга.
flowchart LR
U["Команда<br/>«записать Веру»"] --> A["Агрегат Cohort<br/>свёрнут из журнала"]
A -->|"новые события"| S[("Хранилище событий<br/>журналы потоков")]
S -->|"чтение журнала"| A
S --> P1["Проекция<br/>«свободные места»"]
S --> P2["Проекция<br/>«расписание студента»"]
P1 --> Q1["Витрина каталога"]
P2 --> Q2["Личный кабинет"]Журнал и свёртка
Заголовок раздела «Журнал и свёртка»События одного экземпляра агрегата образуют журнал событий агрегатажурнал событий агрегата (stream)EN event streamУпорядоченная последовательность событий одного экземпляра агрегата; номер события в журнале — его версия. По-английски stream, обычно переводят «поток событий», но в программе DDD слово «поток» занято (Cohort), поэтому говорим «журнал событий потока».. По-английски
это stream, и обычно его переводят «поток событий», но у нас слово «поток» занято: это Cohort. Поэтому говорим
«журнал событий потока python-2026-11» — маленький пример того, как единый язык из
урока DDD.1.2 влияет даже на технические термины.
Номер события в журнале — его версия. Состояние получается свёрткойсвёртка списка (fold, reduce)EN fold, reduceОперация, которая проходит по списку и накапливает результат: начальное значение и функция «состояние + элемент → новое состояние». В Python — functools.reduce или обычный цикл. В event sourcing состояние агрегата — свёртка его событий. Не путать со свёрткой (convolution) в нейросетях. (fold): берём пустое
начальное состояние и по очереди применяем к нему события функцией apply(состояние, событие) → новое состояние.
В Python это functools.reduce(apply, events, None) или обычный цикл. Вот свёртка журнала потока вместимостью 2:
| Версия | Событие | Записи после события | Свободно |
|---|---|---|---|
| 1 | CohortOpened (вместимость 2) | — | 2 |
| 2 | StudentEnrolled (Анна) | Анна — ожидает оплаты | 1 |
| 3 | StudentEnrolled (Борис) | Анна, Борис — ожидают оплаты | 0 |
| 4 | EnrollmentConfirmed (Анна) | Анна — подтверждена, Борис — ожидает | 0 |
| 5 | EnrollmentCancelled (Борис) | Анна — подтверждена, Борис — отменена | 1 |
| 6 | SeatReleased (Борис) | без изменений | 1 |
Событие CohortOpened (ПотокОткрыт: набор на поток открыт) — новое для нашего глоссария. Без него журнал не с чего
начать: вместимость и дата старта тоже должны откуда-то взяться. А SeatReleased идёт сразу за отменой записи,
которая занимала место. Состояния записей оно не меняет, но это отдельный факт для других подписчиков: на него
реагирует политика «предложить место первому в листе ожидания».
Три правила свёртки:
- В
applyнет проверок бизнес-правил. Событие уже случилось, «не разрешить» его нельзя. Правила проверяет команда до того, как записать событие. Если быapplyпроверял вместимость, то после решения экспертов уменьшить вместимость потока старые журналы перестали бы загружаться. - Журнал только дописывают. Ошибку исправляют новым событием, как бухгалтер исправляет проводку сторнирующей записью, а не ластиком. Поэтому журналу можно доверять как аудиту.
- Порядок важен. «Подтверждена» до «записан» — бессмыслица. Версии идут подряд, без пропусков и повторов, и свёртка обязана это проверять. Проверять целостность журнала, а не бизнес-правила.
Хранилище событий
Заголовок раздела «Хранилище событий»Хранилище событийхранилище событийEN event storeБаза данных для event sourcing: хранит события агрегатов в журналах, которые можно только дописывать, и проверяет версию при записи (оптимистическая блокировка). Делается и поверх обычной реляционной базы, и специализированными системами. (event store) умеет немногое: дописать события в журнал и прочитать журнал целиком или начиная с версии. Сохранённое событие кроме данных обычно несёт метаданные: уникальный id, тип, id журнала, версию, время и связи с командой, которая его вызвала (correlation id, causation id) — этот список хорошо разобран в README Оскара Дудыча из источников урока.
Главная деталь — запись с ожидаемой версией: append(cohort_id, events, expected_version). Команда принимала
решение, глядя на журнал версии 6, — значит, дописать можно, только если журнал всё ещё на версии 6. Если кто-то
успел дописать седьмое событие, хранилище отвечает ConcurrencyError: перечитайте журнал и решите заново. Это та же
оптимистическая блокировкаоптимистическая блокировкаEN optimistic locking, optimistic concurrency controlСпособ поймать параллельные изменения без блокировки на время работы: у агрегата есть номер версии, при сохранении хранилище сверяет его с текущим. Версия выросла — кто-то успел раньше, сохранение отклоняется (ConcurrencyError), команду повторяют на свежих данных., что у репозитория из
урока DDD.3.4, и именно она не даёт двум студентам занять одно
последнее место.
Хранилище можно сделать на обычной реляционной базе: таблица events(stream_id, version, type, data, …) с
уникальным ключом (stream_id, version) — вторая вставка той же версии упадёт на ограничении уникальности. Есть и
специализированные базы.
Проекции
Заголовок раздела «Проекции»ПроекцияпроекцияEN projectionКод, который читает события и строит из них модель чтения: таблицу «свободные места по потокам», «расписание студента». Проекцию можно удалить и перестроить заново, прогнав все события с начала. — код, который читает события и строит из них модель чтениямодель чтенияEN read model, query modelДанные, которые нужны человеку или программе, чтобы принять решение: «свободные места на потоках». В Event Storming — зелёный стикер; в CQRS — отдельная модель для чтения (урок DDD.4.1).: «свободные места по потокам» для витрины, «расписание студента» для личного кабинета, «возвраты за месяц» для бухгалтерии. В Event Storming из урока DDD.2.3 модели чтения — зелёные стикеры; здесь видно, откуда они берут данные.
У проекций три важных свойства:
- Их можно перестроить. Модель чтения — производные данные. Удалите таблицу, прогоните все события с начала — получите её заново. Нашли ошибку в проекции — исправили код и перестроили.
- Новую проекцию можно построить задним числом. Бухгалтерия через год попросила отчёт «сколько записей отменяли за сутки до старта»? Журнал хранит всё, что для этого нужно, с первого дня. В обычной базе этих данных уже нет.
- Они должны выдерживать повторы. Событие может прийти в проекцию дважды: после сбоя, перезапуска, повторной доставки. Проекция запоминает, до какой версии каждого журнала она дошла, и пропускает уже учтённое. Это вы реализуете в задании урока.
Проекция обновляется после записи события, поэтому модель чтения отстаёт от агрегата — снова согласованность в конечном счёте.
Если в журнале тысячи событий, сворачивать его на каждую команду долго. Снимокснимок агрегатаEN snapshotСохранённое состояние агрегата на какой-то версии, чтобы не сворачивать весь длинный журнал: загрузка = снимок + хвост событий после него. Это только ускорение: снимок можно выбросить и пересчитать из событий. (snapshot) — сохранённое состояние агрегата на версии N. Загрузка тогда — снимок плюс хвост событий после N:
def load_cohort(store, snapshots, cohort_id): """Загрузить поток: последний снимок (если есть) и события после него.""" snapshot = snapshots.latest(cohort_id) # состояние на версии N или None cohort = Cohort.from_snapshot(snapshot) if snapshot else Cohort() for event in store.load(cohort_id, after_version=cohort.version): cohort._apply(event) return cohortСнимок — только ускорение, а не источник правды. Его можно удалить и пересчитать из событий; при смене кода агрегата старые снимки обычно просто выбрасывают. Проверка корректности простая: снимок плюс хвост обязаны давать то же состояние, что свёртка всего журнала (это проверяет уровень Boss домашней работы).
Прежде чем делать снимки, посмотрите на границы агрегата. Поток курса живёт несколько недель и набирает десятки, от силы сотни событий — снимки ему не нужны. Если журнал растёт бесконечно («все записи школы за все годы»), чаще всего агрегат выбран неудачно. Оскар Дудыч назвал один из своих докладов прямо: «Keep your streams short!» — держите журналы короткими.
Версионирование событий
Заголовок раздела «Версионирование событий»Событие, записанное в журнал, живёт вечно, а код меняется. Допустим, эксперты решили хранить причину отмены записи
(в уроке DDD.2.3 так и сделали: одно событие «ЗаписьОтменена» с причиной).
Старые события EnrollmentCancelled без причины уже лежат в журнале, и переписывать их нельзя — иначе журнал
перестанет быть аудитом. Это задача версионирования событийверсионирование событийEN event versioningКак менять формат событий, которые уже лежат в журнале и которые нельзя переписать: добавлять необязательные поля со значением по умолчанию, вводить новый тип события, переводить старые версии в новую при чтении (upcasting). Грег Янг посвятил теме отдельную книгу «Versioning in an Event Sourced System»., и у неё несколько
решений:
- Новое поле со значением по умолчанию. Старые события читаются как «причина не указана».
- Перевод при чтении (upcasting). Маленькая функция превращает сохранённую версию 1 в версию 2 прямо при загрузке. Остальной код знает только текущую версию.
- Новый тип события. Если смысл факта изменился так, что старое имя вводит в заблуждение, вводят новое событие, а свёртка продолжает понимать и старое.
- Копирование с преобразованием. Переписать все журналы в новое хранилище в новом формате. Тяжело и рискованно, к этому прибегают редко.
Так выглядит upcasting для событий, сохранённых в виде словарей (как они лежат в JSON):
def upcast(raw): """Привести сохранённое событие к текущей версии схемы — при чтении, не трогая журнал.""" if raw["type"] == "EnrollmentCancelled" and raw["schema"] == 1: # в версии 2 появилась причина отмены; у старых событий её нет — так и пишем raw = {**raw, "schema": 2, "data": {**raw["data"], "reason": "не указана"}} return raw
old = {"type": "EnrollmentCancelled", "schema": 1, "data": {"cohort_id": "python-2026-11", "student_id": "boris"}}new = {"type": "EnrollmentCancelled", "schema": 2, "data": {"cohort_id": "python-2026-11", "student_id": "vera", "reason": "не оплачена за 48 часов"}}for raw in [old, new]: print(upcast(raw)["data"]){'cohort_id': 'python-2026-11', 'student_id': 'boris', 'reason': 'не указана'}{'cohort_id': 'python-2026-11', 'student_id': 'vera', 'reason': 'не оплачена за 48 часов'}Грег Янг написал об этой теме целую книгу — «Versioning in an Event Sourced System». Главный совет из практики: думайте о формате события как о публичном контракте с будущими версиями своего же кода.
Плюсы и цена
Заголовок раздела «Плюсы и цена»| Что даёт | Чего стоит |
|---|---|
| Полный аудит: на вопрос «почему Борис потерял место» отвечает журнал, а не догадки | Сложность: непривычная модель, новые понятия, новые ошибки |
| Состояние на любой момент в прошлом | Версионирование событий: прошлое нельзя «смигрировать» |
| Новые проекции и отчёты задним числом | Модели чтения отстают от агрегатов, интерфейс должен это учитывать |
| Отладка повтором: прогнать журнал через исправленный код | Произвольный запрос «по всем данным» — только через проекцию |
| Никаких затираний: обновление не уничтожает прежнее значение | Удалить персональные данные из неизменяемого журнала трудно — нужны особые приёмы (например, хранить их отдельно от событий) |
| События уже готовы для других контекстов (следующий урок) | Команде нужно время и опыт; ошибки дорого обходятся годами |
Последнюю строку стоит принять всерьёз: даже у Грега Янга есть доклад «Why Event Sourced Systems Fail».
Где это уместно в школе
Заголовок раздела «Где это уместно в школе»Event sourcing — решение для конкретного контекста или даже агрегата, а не для всей системы.
- Набор — хороший кандидат. Ядро школы, много споров «кто занял место» и «почему мне не вернули деньги», история записей ценна для бизнеса, журналы потоков короткие.
- Оплата по природе похожа на журнал (платежи и возвраты), но это типовой поддомен: её покупают, а не пишут.
- Каталог — CRUD, event sourcing здесь только навредит.
- Обучение — возможно, для прогресса студента: «как он шёл по курсу» интересно методистам.
Простая проверка: если бизнес часто спрашивает «как мы сюда попали?» и история имеет ценность — event sourcing стоит обсудить. Если важно только «что сейчас» — храните состояние, публикуйте доменные события из урока DDD.3.3 и надёжно доставляйте их другим контекстам — как, разберём в следующем уроке. Доменные события и event sourcing — не одно и то же: первые можно публиковать и без второго.
Практика
Заголовок раздела «Практика»Поток из событий: учебная реализация
Заголовок раздела «Поток из событий: учебная реализация»Соберём всё вместе: агрегат Cohort, который получает состояние свёрткой событий, хранилище событий в памяти с
проверкой версии, сервис приложения и проекцию «расписание студента». Имена и правила — из глоссария школы, но
агрегат проще, чем в уроке DDD.3.3: лист ожидания и деньги опущены, записи различаются по student_id без
отдельного номера, а у события отмены нет причины и суммы возврата — всё ради того, чтобы код поместился на экран.
Сохраните в es_cohort.py и запустите.
from dataclasses import dataclassfrom datetime import datetimefrom enum import Enum
class EnrollmentStatus(Enum): PENDING_PAYMENT = "ожидает оплаты" CONFIRMED = "подтверждена" CANCELLED = "отменена"
class NoFreeSeats(Exception): """На потоке нет свободных мест."""
class EnrollmentClosed(Exception): """Запись на поток закрыта: поток уже стартовал."""
class AlreadyEnrolled(Exception): """Студент уже записан на этот поток."""
class ConcurrencyError(Exception): """Журнал потока изменился, пока мы принимали решение."""
# ---------- События: неизменяемые факты в прошедшем времени ----------@dataclass(frozen=True)class CohortOpened: cohort_id: str capacity: int starts_at: datetime
@dataclass(frozen=True)class StudentEnrolled: cohort_id: str student_id: str enrolled_at: datetime
@dataclass(frozen=True)class EnrollmentConfirmed: cohort_id: str student_id: str
@dataclass(frozen=True)class EnrollmentCancelled: cohort_id: str student_id: str
@dataclass(frozen=True)class SeatReleased: cohort_id: str student_id: str
# ---------- Агрегат: решает по правилам, а состояние получает из событий ----------class Cohort: """Поток курса, состояние которого — свёртка его событий."""
def __init__(self): self.cohort_id, self.capacity, self.starts_at = None, 0, None self.statuses = {} # student_id → EnrollmentStatus self.version = 0 # сколько событий учтено self._new_events = [] # решения этой команды, ещё не сохранённые
@classmethod def from_events(cls, history): """Восстановить поток: применить события его журнала по порядку.""" cohort = cls() for event in history: cohort._apply(event) return cohort
def _apply(self, event): """Шаг свёртки. Правил здесь нет: событие уже случилось, его нельзя «не разрешить».""" if isinstance(event, CohortOpened): self.cohort_id, self.capacity, self.starts_at = event.cohort_id, event.capacity, event.starts_at elif isinstance(event, StudentEnrolled): self.statuses[event.student_id] = EnrollmentStatus.PENDING_PAYMENT elif isinstance(event, EnrollmentConfirmed): self.statuses[event.student_id] = EnrollmentStatus.CONFIRMED elif isinstance(event, EnrollmentCancelled): self.statuses[event.student_id] = EnrollmentStatus.CANCELLED elif not isinstance(event, SeatReleased): # SeatReleased состояние не меняет raise TypeError(f"Неизвестное событие {type(event).__name__}") self.version += 1
def _record(self, event): """Принятое решение: применить событие к себе и запомнить как новое.""" self._apply(event) self._new_events.append(event)
def pull_events(self): """Отдать новые события и забыть их (их сохранит сервис приложения).""" events, self._new_events = self._new_events, [] return events
# --- запрос --- def free_seats(self): taking = [s for s in self.statuses.values() if s is not EnrollmentStatus.CANCELLED] return self.capacity - len(taking)
# --- команды: проверить правила, затем записать событие --- @classmethod def open(cls, cohort_id, capacity, starts_at): cohort = cls() cohort._record(CohortOpened(cohort_id, capacity, starts_at)) return cohort
def enroll(self, student_id, now): if now >= self.starts_at: raise EnrollmentClosed(f"Запись на поток {self.cohort_id} закрыта") if self.statuses.get(student_id, EnrollmentStatus.CANCELLED) is not EnrollmentStatus.CANCELLED: raise AlreadyEnrolled(f"{student_id} уже записан на поток {self.cohort_id}") if self.free_seats() == 0: raise NoFreeSeats(f"На потоке {self.cohort_id} нет свободных мест") self._record(StudentEnrolled(self.cohort_id, student_id, now))
def confirm(self, student_id): # упрощение: подтвердить можно только ожидающую оплаты запись, иначе ничего не происходит if self.statuses.get(student_id) is EnrollmentStatus.PENDING_PAYMENT: self._record(EnrollmentConfirmed(self.cohort_id, student_id))
def cancel(self, student_id): if self.statuses.get(student_id) in (EnrollmentStatus.PENDING_PAYMENT, EnrollmentStatus.CONFIRMED): self._record(EnrollmentCancelled(self.cohort_id, student_id)) self._record(SeatReleased(self.cohort_id, student_id))
# ---------- Хранилище событий: журналы только дописываются ----------class InMemoryEventStore: def __init__(self): self.streams = {} # cohort_id → журнал событий потока self.log = [] # все события по порядку — их читают проекции
def load(self, cohort_id): return list(self.streams.get(cohort_id, []))
def append(self, cohort_id, events, expected_version): stream = self.streams.setdefault(cohort_id, []) if len(stream) != expected_version: raise ConcurrencyError(f"Журнал {cohort_id}: ждали версию {expected_version}, " f"а там уже {len(stream)}") stream.extend(events) self.log.extend(events)
def execute(store, cohort_id, command): """Сервис приложения: загрузить поток → выполнить команду → дописать новые события.""" cohort = Cohort.from_events(store.load(cohort_id)) command(cohort) new = cohort.pull_events() store.append(cohort_id, new, expected_version=cohort.version - len(new))
# ---------- Проекция: модель чтения «расписание студента» ----------class StudentSchedule: def __init__(self): self.starts = {} # cohort_id → старт потока (запоминаем из CohortOpened) self.rows = {} # student_id → {cohort_id: статус по-русски}
def handle(self, event): if isinstance(event, CohortOpened): self.starts[event.cohort_id] = event.starts_at elif isinstance(event, StudentEnrolled): self.rows.setdefault(event.student_id, {})[event.cohort_id] = "ожидает оплаты" elif isinstance(event, EnrollmentConfirmed): self.rows[event.student_id][event.cohort_id] = "подтверждена" elif isinstance(event, EnrollmentCancelled): self.rows[event.student_id][event.cohort_id] = "отменена"
def for_student(self, student_id): rows = self.rows.get(student_id, {}) return sorted((self.starts[c], c, status) for c, status in rows.items())
store = InMemoryEventStore()for cohort in [Cohort.open("python-2026-11", 2, datetime(2026, 11, 2, 10, 0)), Cohort.open("ds-2026-12", 2, datetime(2026, 12, 7, 10, 0))]: store.append(cohort.cohort_id, cohort.pull_events(), expected_version=0)
now = datetime(2026, 10, 1, 12, 0)execute(store, "ds-2026-12", lambda c: c.enroll("anna", now))execute(store, "python-2026-11", lambda c: c.enroll("anna", now))execute(store, "python-2026-11", lambda c: c.enroll("boris", now))execute(store, "python-2026-11", lambda c: c.confirm("anna"))execute(store, "python-2026-11", lambda c: c.cancel("boris"))execute(store, "ds-2026-12", lambda c: c.confirm("anna"))
print("Журнал потока python-2026-11:")for version, event in enumerate(store.load("python-2026-11"), start=1): print(f" {version}. {type(event).__name__}")python = Cohort.from_events(store.load("python-2026-11"))print("Состояние:", {s: st.value for s, st in python.statuses.items()}, "| свободно мест:", python.free_seats(), "| версия:", python.version)
# Два запроса одновременно прочитали версию 6 и хотят занять последнее местоfirst = Cohort.from_events(store.load("python-2026-11"))second = Cohort.from_events(store.load("python-2026-11"))first.enroll("vera", now)second.enroll("gleb", now)store.append("python-2026-11", first.pull_events(), expected_version=6)try: store.append("python-2026-11", second.pull_events(), expected_version=6)except ConcurrencyError as error: print("Второй запрос:", error)
schedule = StudentSchedule()for event in store.log: # проекцию можно построить с нуля в любой момент schedule.handle(event)print("Расписание Анны:")for starts_at, cohort_id, status in schedule.for_student("anna"): print(f" {starts_at:%d.%m.%Y} {cohort_id:15} {status}")Журнал потока python-2026-11: 1. CohortOpened 2. StudentEnrolled 3. StudentEnrolled 4. EnrollmentConfirmed 5. EnrollmentCancelled 6. SeatReleasedСостояние: {'anna': 'подтверждена', 'boris': 'отменена'} | свободно мест: 1 | версия: 6Второй запрос: Журнал python-2026-11: ждали версию 6, а там уже 7Расписание Анны: 02.11.2026 python-2026-11 подтверждена 07.12.2026 ds-2026-12 подтвержденаНа что посмотреть:
- Команда и событие разделены.
enrollпроверяет правила (запись закрыта, повторная запись, нет мест) и только потом вызывает_record._applyне проверяет ничего — он переносит факт в состояние. Тот же_applyиспользуется и при восстановлении из журнала, поэтому состояние после команды и после перезагрузки совпадает. pull_events()— как в уроке DDD.3.3. Агрегат копит решения, сервис приложения забирает их и сохраняет. Разница с обычным репозиторием в том, что сохраняются только события, а не состояние.- Гонка за последнее место решена версией. Оба запроса увидели одно свободное место, оба
enrollпрошли проверки. Но дописать события на версию 6 удалось только первому; второй получилConcurrencyError. Сервис приложения в этом случае перечитает журнал, повторит команду — иenrollчестно ответитNoFreeSeats. - Проекция ничего не знает об агрегате.
StudentScheduleчитает общий журнал и строит данные под экран личного кабинета: потоки разных курсов одного студента, отсортированные по дате старта. Чтобы показать дату старта, проекция сама запоминает её изCohortOpened— у моделей чтения бывают свои вспомогательные данные.
Задание: свёртка и проекция «свободные места»
Заголовок раздела «Задание: свёртка и проекция «свободные места»»Теперь сами. В задании события несут номер версии, и ваш код должен следить за порядком. Напишите:
apply_event(state, event)— шаг свёртки с проверкой целостности журнала: начало сCohortOpened, версии подряд, свой поток, событие о существующей записи;rebuild(events)— состояние потока свёрткой всего журнала;FreeSeatsProjection.handle(event)— проекцию «сколько свободных мест по потокам», которая получает журналы всех потоков вперемешку, переживает повторную доставку и не перескакивает через пропущенное событие.
Неизвестный тип события и в свёртке, и в проекции — ошибка UnknownEventType, а не тихий пропуск: событие, которое
код не понимает, почти всегда означает, что код отстал от журнала.
from dataclasses import dataclass, field
from datetime import datetime
from functools import reduce # пригодится в rebuild
# Статусы записи на поток — как в глоссарии школы (урок DDD.1.2)
PENDING_PAYMENT = "PENDING_PAYMENT" # ожидает оплаты
CONFIRMED = "CONFIRMED" # подтверждена
CANCELLED = "CANCELLED" # отменена
class EventOrderError(Exception):
"""Журнал событий нарушен: чужой поток, пропуск версии, событие о записи, которой нет."""
class UnknownEventType(Exception):
"""Событие неизвестного типа: молча пропустить его нельзя."""
# События потока. У каждого: cohort_id — чей это журнал, version — номер события в журнале (1, 2, 3, …)
@dataclass(frozen=True)
class CohortOpened:
"""ПотокОткрыт: набор на поток открыт. С этого события начинается журнал потока."""
cohort_id: str
version: int
capacity: int
starts_at: datetime
@dataclass(frozen=True)
class StudentEnrolled:
"""СтудентЗаписан: запись создана, ожидает оплаты и занимает место."""
cohort_id: str
version: int
student_id: str
@dataclass(frozen=True)
class EnrollmentConfirmed:
"""ЗаписьПодтверждена: платёж получен."""
cohort_id: str
version: int
student_id: str
@dataclass(frozen=True)
class EnrollmentCancelled:
"""ЗаписьОтменена: студент отменил запись или 48 часов прошли без оплаты."""
cohort_id: str
version: int
student_id: str
@dataclass(frozen=True)
class SeatReleased:
"""МестоОсвободилось: идёт сразу за отменой записи, которая занимала место."""
cohort_id: str
version: int
student_id: str
EVENT_TYPES = (CohortOpened, StudentEnrolled, EnrollmentConfirmed, EnrollmentCancelled, SeatReleased)
@dataclass
class CohortState:
"""Состояние потока, восстановленное из событий."""
cohort_id: str
capacity: int
starts_at: datetime
version: int = 0
enrollments: dict = field(default_factory=dict) # student_id → статус записи
def free_seats(self):
"""Вместимость минус записи «ожидает оплаты» и «подтверждена»."""
taking_seats = [s for s in self.enrollments.values() if s != CANCELLED]
return self.capacity - len(taking_seats)
def apply_event(state, event):
"""Один шаг свёртки: состояние до события + событие → состояние после.
Порядок проверок:
1) тип события не из EVENT_TYPES → UnknownEventType (молча пропускать нельзя);
2) CohortOpened: допустим, только если state is None и version == 1, иначе EventOrderError;
вернуть новый CohortState(cohort_id, capacity, starts_at, version=1);
3) любое другое событие при state is None → EventOrderError (журнал начинается с CohortOpened);
чужой cohort_id или version != state.version + 1 → EventOrderError;
4) StudentEnrolled → enrollments[student_id] = PENDING_PAYMENT;
EnrollmentConfirmed / EnrollmentCancelled / SeatReleased о студенте, которого нет в
enrollments, → EventOrderError; Confirmed → CONFIRMED, Cancelled → CANCELLED;
SeatReleased статусы не меняет (место уже вернула отмена);
5) state.version = event.version; вернуть state (менять его на месте можно).
"""
# TODO: ваш код
...
def rebuild(events):
"""Состояние потока из его журнала: свёртка apply_event, начиная с None.
Подойдёт functools.reduce(apply_event, events, None) или обычный цикл.
Пустой журнал → EventOrderError.
"""
# TODO: ваш код
...
class FreeSeatsProjection:
"""Модель чтения «свободные места по потокам» — для витрины каталога."""
def __init__(self):
self.free = {} # cohort_id → свободных мест
self.versions = {} # cohort_id → версия последнего учтённого события
def handle(self, event):
"""Учесть одно событие из общей шины, где перемешаны журналы всех потоков.
1) тип не из EVENT_TYPES → UnknownEventType;
2) last = self.versions.get(event.cohort_id, 0);
event.version <= last → уже учтено, просто выйти (идемпотентность);
event.version > last + 1 → EventOrderError (событие пропущено);
3) CohortOpened → free[cohort_id] = capacity; StudentEnrolled → минус одно место;
SeatReleased → плюс одно; Confirmed и Cancelled места не меняют;
4) запомнить версию: self.versions[cohort_id] = event.version.
"""
# TODO: ваш код
...
def seats(self, cohort_id):
"""Сколько свободных мест на потоке."""
return self.free[cohort_id]
# Журналы двух потоков вперемешку — так события приходят из общей шины
PY, DS = "python-2026-11", "ds-2026-12"
HISTORY = [
CohortOpened(PY, 1, capacity=3, starts_at=datetime(2026, 11, 2, 10, 0)),
CohortOpened(DS, 1, capacity=2, starts_at=datetime(2026, 12, 7, 10, 0)),
StudentEnrolled(PY, 2, "anna"),
StudentEnrolled(PY, 3, "boris"),
StudentEnrolled(DS, 2, "vera"),
EnrollmentConfirmed(PY, 4, "anna"),
EnrollmentCancelled(PY, 5, "boris"),
SeatReleased(PY, 6, "boris"),
StudentEnrolled(DS, 3, "gleb"),
]
python_cohort = rebuild([e for e in HISTORY if e.cohort_id == PY])
if python_cohort is not None:
print("Поток", python_cohort.cohort_id, "версия", python_cohort.version)
print(" записи:", python_cohort.enrollments)
print(" свободных мест:", python_cohort.free_seats())
projection = FreeSeatsProjection()
for event in HISTORY + HISTORY[:4]: # хвост пришёл повторно
projection.handle(event)
print("Проекция «свободные места»:", projection.free)
Обратите внимание на тест с перемешанными журналами: проекция получает события трёх потоков вперемешку, как из общей шины, и всё равно совпадает со свёрткой каждого потока. Порядок важен только внутри одного журнала — между журналами разных агрегатов его нет и быть не должно.
Проверь себя
Заголовок раздела «Проверь себя»- CQS — правило для методов: либо меняй, либо отвечай. CQRS (Грег Янг, около 2010 года) — разные модели для записи и для чтения. Это шкала: от отдельных запросов к тем же таблицам до отдельных хранилищ, которые обновляют обработчики событий. Ни вторая база, ни event sourcing не обязательны.
- Чтение может отставать, запись — никогда: агрегат проверяет правила по актуальным данным, а модели чтения согласованы в конечном счёте. В CRUD-частях системы CQRS лишний.
- Event sourcing хранит события вместо состояния. Состояние — свёртка журнала функцией
apply, в которой нет бизнес-правил, только проверка целостности. Хранилище событий дописывает журнал с проверкой ожидаемой версии. - Проекции строят модели чтения из событий, их можно перестроить и создать задним числом; они обязаны выдерживать повторную доставку. Снимки — только ускорение. Формат событий меняют без переписывания журнала: значения по умолчанию, upcasting, новые типы.
- Плата — сложность, версионирование и отставание моделей чтения. В школе event sourcing уместен в Наборе, но не в Каталоге; решать по контексту, а не для всей системы.
Дальше: урок DDD.4.2 — интеграция контекстов. События выходят за границу Набора: чем интеграционное событие отличается от доменного, как надёжно опубликовать событие через outbox, почему получатель должен быть идемпотентным и как провести процесс «оплата прошла → запись подтверждена → сертификат выдан» через несколько контекстов. Там же — домашняя работа ddd-hw02-integration.