Доменные события, сервисы и фабрики
Ещё три строительных блока тактического DDD и сценарий, который их связывает: доменные события как факты в прошедшем времени, доменные сервисы для правил, которым не место в сущностях, фабрики для сложного создания агрегата и прикладной сервис без бизнес-правил — на полном сценарии «записаться на поток».
Зачем это нужно
Заголовок раздела «Зачем это нужно»В уроке DDD.3.2 агрегат «Поток» научился соблюдать правила Набора: не больше записей, чем мест, никаких повторных записей, запись закрывается в момент старта. Но запись на поток — не только строчка в списке. Оплате нужно выставить счёт, Уведомлениям — отправить письмо, освободившееся место надо предложить листу ожидания. Куда писать этот код? Первое желание — прямо в агрегат:
def enroll(self, student_id, now): ... # правила записи payments_api.create_invoice(student_id, self.price) # HTTP-запрос к платёжному модулю smtp.send(student_id, "Вы записаны на поток") # письмоТеперь модель зависит от HTTP и почтового сервера, тест записи шлёт настоящие письма, а каждая новая реакция правит агрегат. Хуже того: если сохранение сорвётся, письмо уже ушло — о записи, которой нет. Писать реакции в обработчиках веб-запросов не лучше: логика разъедется по личному кабинету, поддержке и импорту, и получится история из урока DDD.1.1 — два кода отмены, два разных возврата.
Есть и другие вопросы. Где жить правилу о скидках, которому нужны данные и потока, и Сертификации? Кто проверяет, что поток не открывают со стартом в прошлом, и кто выдаёт ему номер? Ответ тактического DDD — доменные события, доменные сервисы, фабрики и прикладной сервис, который связывает их в сценарий «записаться на поток».
Доменные события
Заголовок раздела «Доменные события»Факт, который важен экспертам
Заголовок раздела «Факт, который важен экспертам»Доменное событиедоменное событиеEN domain eventЗначимый для экспертов факт, который уже произошёл, в прошедшем времени: «Студент записан», «Платёж получен». В Event Storming — оранжевый стикер, в коде — неизменяемый объект (урок DDD.3.3). (domain event) — неизменяемая запись о том, что в предметной области случилось нечто важное для экспертов: «студент записан», «запись подтверждена», «место освободилось». На Event Storming из урока DDD.2.3 это оранжевые стикеры, и таблица «стикер → код» в том уроке уже наметила путь: команда становится методом агрегата или прикладного сервиса, событие — классом, политика («когда место освободилось — предложить место») — обработчиком события. Этот урок проходит путь в коде. В книге Эванса 2003 года шаблона событий ещё нет — его выработало сообщество позже; у Вона Вернона (2013) событиям посвящена глава 8, а Эванс добавил их в справочник DDD Reference (2015).
У события три свойства:
- Прошедшее время. Событие уже случилось. «Отменить» запись — значит совершить новое действие со своим
событием
EnrollmentCancelled. - Неизменяемость. Историю не переписывают:
@dataclass(frozen=True). Авторы «Cosmic Python» называют событие разновидностью объекта-значения из урока DDD.3.1. - Язык экспертов. Методист читает
StudentEnrolledкак «СтудентЗаписан» из глоссария.
Событие легко спутать с командой и с отказом:
| Команда | Событие | Отказ | |
|---|---|---|---|
| Что это | просьба «сделай» | факт «случилось» | «не сделаю: нарушено правило» |
| Форма | глагол: cohort.enroll(...) | прошедшее время: StudentEnrolled | исключение: NoFreeSeats |
| Может не случиться? | да, агрегат вправе отказать | нет, уже случилось | ничего не изменилось |
| Кому адресовано | одному агрегату | любым подписчикам; агрегат их не знает | тому, кто прислал команду |
Отказ в нашей модели — не событие: при NoFreeSeats ничего не изменилось. «Cosmic Python» советует не описывать одно
понятие и исключением, и событием сразу.
Как называть события
Заголовок раздела «Как называть события»Имя события — подлежащее и глагол прошедшего времени из глоссария урока DDD.1.2; одно событие — один факт.
| Плохо | Почему | Лучше |
|---|---|---|
EnrollStudent | это команда, а не факт | StudentEnrolled |
CohortUpdated | «что-то поменялось» — подписчику придётся гадать что | StudentEnrolled, SeatReleased |
EnrollmentStatusChanged(status=2) | техника вместо смысла, статус числом | EnrollmentConfirmed |
EmailSent в Наборе | факт Уведомлений, а не Набора | событие живёт в своём контексте |
Что хранить в событии
Заголовок раздела «Что хранить в событии»В событие кладут то, что нужно подписчикам, чтобы отреагировать, не возвращаясь к агрегату с вопросами:
- идентификаторы, а не сами объекты: ссылку на живой агрегат подписчик может изменить, да и состояние к моменту обработки уже другое;
- момент
occurred_at— тот жеnow, что пришёл в команду; - факты на момент события, которые потом могут измениться: цену записи, срок оплаты, сумму возврата.
В EnrollmentCancelled лежит refund — сумма возврата, решённая агрегатом по
правилу 7 суток. Оплата не пересчитывает правило, а просто возвращает refund: правило живёт в одном месте. Там же
причина reason: на Event Storming эксперты решили, что отмена студентом и отмена по таймеру 48 часов — одно
событие с разной причиной, а не два.
Чего в событии быть не должно: агрегата целиком, данных «на всякий случай», персональных данных, которые подписчик
знает по student_id. Событие — контракт с подписчиками: переименовать поле — сломать чужой код. Начинайте с
минимума, а о версиях событий поговорим в уроке DDD.4.1.
События в коде агрегата
Заголовок раздела «События в коде агрегата»В этом уроке у Набора четыре события — StudentEnrolled, EnrollmentConfirmed, EnrollmentCancelled,
SeatReleased (файл events.py — в практике); лист ожидания добавит SeatOffered и SeatOfferExpired, а у Оплаты
свои — InvoiceIssued, PaymentRefunded. Агрегат не рассылает события сам: он складывает их в список, а забирает
их тот, кто ведёт сценарий:
class Cohort: def enroll(self, student_id, now): ... # проверки правил — как в уроке DDD.3.2 self._events.append(StudentEnrolled( self.cohort_id, enrollment_id, student_id, price=self.price, expires_at=now + PAYMENT_PERIOD, occurred_at=now)) return enrollment_id
def pull_events(self): """Отдать накопленные события и очистить список: каждое событие уходит один раз.""" events, self._events = self._events, [] return eventsПочему не bus.publish(...) прямо в enroll? Домен остаётся без инфраструктуры: тест агрегата проверяет
pull_events() и не настраивает шин. Момент публикации выбирает тот, кто знает о транзакции. А события одной
команды (cancel даёт EnrollmentCancelled и SeatReleased) уходят вместе. В руководстве Microsoft по
микросервисам этот приём называют отложенным, а канва агрегата DDD Crew (Aggregate Design Canvas) так и записывает
пары «команда → события».
Публиковать — после сохранения
Заголовок раздела «Публиковать — после сохранения»| Когда публиковать | Что может пойти не так |
|---|---|
| до сохранения агрегата | сохранение сорвалось (конфликт, база недоступна) — а счёт уже выставлен и письмо ушло о записи, которой нет |
| после сохранения | процесс упал между сохранением и публикацией — запись есть, а счёт никто не выставил |
Второй вариант лучше: «событие потерялось» исправляет надёжная доставка, а «мир узнал о том, чего не было» — уже
ничто. Надёжная доставка — шаблон outboxtransactional outbox (исходящие сообщения в той же транзакции)EN Transactional OutboxШаблон из каталога microservices.io Криса Ричардсона: события пишут в таблицу outbox в той же транзакции, что и изменение агрегата, а отдельный процесс-ретранслятор (relay) потом публикует их в брокер и помечает отправленными. Даёт доставку «как минимум один раз».: события сохраняют в одной транзакции
с агрегатом, а публикует их отдельный шаг. Это тема урока DDD.4.2, там же —
отличие доменных событий (внутри контекста) от интеграционных (между контекстами). Здесь мы ради простоты передаём
StudentEnrolled прямо в Оплату.
Доменные сервисы
Заголовок раздела «Доменные сервисы»Когда операции нет места в сущности
Заголовок раздела «Когда операции нет места в сущности»Эванс замечает: некоторые важные операции не принадлежат естественным образом ни одной сущности и ни одному объекту-значению. Втиснуть такую операцию в объект — исказить его; завести объект «Менеджер» — получить жаргон. Для них есть доменный сервисдоменный сервисEN domain serviceОперация предметной области без состояния, которая не принадлежит естественным образом ни одной сущности или объекту-значению; интерфейс выражен через элементы модели (Эванс). Пример — цена записи со скидками или перевод между счетами. Сначала правило пробуют поместить в агрегат, сервис — последнее средство. (domain service). У хорошего сервиса три признака:
- операция — понятие предметной области, но не естественная часть сущности или объекта-значения;
- интерфейс выражен через элементы модели: потоки, деньги, периоды, а не словари и строки базы;
- у операции нет состояния: результат зависит только от аргументов.
Классический пример Эванса — перевод денег между счетами: в нём участвуют два счёта, и банкир говорит именно «перевод». Эванс различает сервисы прикладные (ведут сценарий), доменные (бизнес-правила) и инфраструктурные (отправить письмо). Здесь — доменные.
Пример: цена записи со скидками
Заголовок раздела «Пример: цена записи со скидками»В уроке DDD.1.2 линтер нашёл слово discount, которого нет в
глоссарии, и мы решили спросить экспертов. Допустим, они ответили (условия учебного примера): 5 % за запись за 30
суток и более до старта, 10 % выпускнику, у которого уже есть сертификат школы; скидки не суммируются, действует
большая. В глоссарий добавляем «скидку» и «выпускника». Это правило учебной модели; в
уроке DDD.4.3 вы увидите, что старый код школы считал раннюю скидку иначе, — и
разберёте, что делать с таким расхождением.
В Cohort правило не ложится: поток не знает, есть ли у студента сертификат, — это данные Сертификации; к тому же
цены меняет маркетинг, а правила записи — методисты. В Student — некуда: в Наборе студент — просто student_id.
Остаётся доменный сервис — обычная функция:
"""Доменный сервис «Цена записи»: правило о скидках, которому не место ни в одной сущности."""from datetime import timedeltafrom decimal import Decimal
EARLY_ENROLLMENT_PERIOD = timedelta(days=30) # условия учебного примера:EARLY_ENROLLMENT_DISCOUNT = 5 # запись за 30 суток и более до старта — минус 5 %,ALUMNI_DISCOUNT = 10 # выпускнику (есть сертификат школы) — минус 10 %
def enrollment_price(cohort, now, is_alumnus): """Цена записи на поток для студента. Скидки не суммируются — действует большая.""" discount = 0 if cohort.starts_at - now >= EARLY_ENROLLMENT_PERIOD: discount = EARLY_ENROLLMENT_DISCOUNT if is_alumnus: discount = max(discount, ALUMNI_DISCOUNT) return cohort.price * (Decimal(100 - discount) / 100) # округление — как в Money: до копейки# demo_pricing.py — запускать из корня проекта (файлы пакета enrollment — в практике)from datetime import datetime
from enrollment.model import Cohort, Coursefrom enrollment.money import Moneyfrom enrollment.pricing import enrollment_price
python = Course("python", price=Money(3_000_000, "RUB")) # 30 000 руб.cohort = Cohort.open(python, datetime(2026, 11, 2, 10, 0), capacity=30, now=datetime(2026, 9, 1))
for now, is_alumnus in [(datetime(2026, 9, 1), False), (datetime(2026, 9, 1), True), (datetime(2026, 10, 20), False)]: days = (cohort.starts_at - now).days who = "выпускник" if is_alumnus else "новичок " print(f"за {days:2} сут. до старта, {who}: {enrollment_price(cohort, now, is_alumnus)}")за 62 сут. до старта, новичок : 28 500,00 RUBза 62 сут. до старта, выпускник: 27 000,00 RUBза 13 сут. до старта, новичок : 30 000,00 RUBВсе три признака на месте: слово экспертов, вход и выход — элементы модели, состояния нет. В Сертификацию сервис
не ходит: is_alumnus ему передаёт прикладной сервис, спросив соседний контекст через порт
(урок DDD.3.4), поэтому тест сервиса — таблица примеров без фейков.
Осторожно: сервисы затягивают. Заведите EnrollmentRules.can_enroll(cohort, student_id) с проверкой вместимости — и
поток станет мешком с данными. Сначала пробуйте поместить правило в агрегат или объект-значение; сервис — только если
оно не принадлежит ни одному из них. Завись скидка только от данных потока, ей место в методе Cohort.
Фабрики
Заголовок раздела «Фабрики»Сложное рождение
Заголовок раздела «Сложное рождение»Открыть поток — не просто вызвать конструктор: нужно проверить правила рождения (старт — в будущем), выдать номер, взять цену у курса и собрать агрегат так, чтобы инварианты соблюдались с первой секунды. Если этим займётся каждый клиент — панель методиста, импорт расписания, тесты, — правила продублируются. Эванс выносит сложное создание в фабрикуфабрикаEN factoryШаблон DDD для сложного создания: собирает агрегат целиком и сразу корректным, проверяет правила рождения и выдаёт номера. Формы — фабричный метод корня (cohort.enroll создаёт запись), отдельная фабрика (Cohort.open) и восстановление из хранилища. (factory): она создаёт агрегат целиком и сразу корректным.
Три формы фабрики
Заголовок раздела «Три формы фабрики»Фабричный метод корня. Вы его уже видели: cohort.enroll(...) создаёт Enrollment внутри агрегата. Снаружи
никто не пишет Enrollment(...) — номер и проверки достаются корню.
Отдельная фабрика — для самого агрегата. В Python это метод класса или функция; класс CohortFactory нужен,
только если у фабрики свои зависимости, например генератор номеров.
@classmethod def open(cls, course, starts_at, capacity, now): """Фабрика: открыть новый поток курса. Здесь — правила рождения и выдача номера.""" if starts_at <= now: raise ValueError(f"Нельзя открыть поток со стартом в прошлом: {starts_at:%d.%m.%Y %H:%M}") cohort_id = f"{course.slug}-{starts_at:%Y-%m}" # номер по правилам школы: курс-год-месяц return cls(cohort_id, starts_at, capacity, price=course.price)Здесь же ответ на вопрос урока DDD.3.1 «кто выдаёт номера». Номер потока — по правилам бизнеса: курс и месяц старта
(в учебной школе у курса не больше одного потока в месяц). Номера записей выдаёт сам поток: python-2026-11-1,
python-2026-11-2 — в тестах они предсказуемы. Если номер должен быть уникален во всей системе, фабрика получает
генератор снаружи; Вернон предлагает просить следующий номер у репозитория.
Course — курс глазами Набора, только адрес и цену (урок DDD.2.1).
Фабрика копирует цену в поток; меняется ли она вслед за Каталогом — вопрос к экспертам.
Восстановление из хранилища (reconstitution). Поток из базы не рождается — он уже существует. Поэтому
восстановление не проверяет правила рождения (поток, стартовавший месяц назад, загружается, хотя open его бы не
открыл) и не порождает событий: ничего не случилось, мы только прочитали. У нас восстановление — это конструктор
Cohort(...): он принимает готовое состояние и проверяет лишь то, что верно всегда (вместимость не меньше 1).
Фабрика может и записать событие о рождении — в уроке DDD.4.1 журнал потока начнётся с «ПотокОткрыт»
(CohortOpened). А простому созданию фабрика не нужна: Money(3_000_000, "RUB") справляется сам.
Прикладной сервис
Заголовок раздела «Прикладной сервис»Сценарий использования
Заголовок раздела «Сценарий использования»Прикладной сервисприкладной сервис (сервис приложения)EN application serviceТочка входа для одного сценария использования: загрузить агрегат, вызвать доменный метод, сохранить, опубликовать события. Тонкий слой без бизнес-правил: знает о хранилище, часах и транзакциях только через порты. (application service, сервис приложения) — точка входа в модель для одного сценария использования: «записаться на поток», «подтвердить запись», «отменить запись». По Эвансу прикладной слой определяет, какую работу выполняет программа, но остаётся тонким: бизнес-правил в нём нет, он координирует задачи и делегирует работу объектам модели. Типичные шаги (у «Cosmic Python» почти те же):
def enroll(self, cohort_id, student_id, now): # 1. команда: кто, что и когда """Сценарий «записаться на поток».""" cohort = self._cohorts.get(cohort_id) # 2. загрузить агрегат cohort.enroll(student_id, now) # 3. команда агрегату — правила там self._cohorts.save(cohort) # 4. сохранить events = cohort.pull_events() # 5. забрать события… self._publisher.publish(events) # …и опубликовать после сохранения return events # 6. вернуть результатПрава доступа, журнал, транзакция, повтор при конфликте — тоже его забота. А if о вместимости, сроках и
возвратах — нет.
sequenceDiagram actor S as Студент participant W as Сайт школы participant A as EnrollmentService participant R as Хранилище потоков participant C as Cohort participant B as Шина событий S->>W: «Записаться на поток» W->>A: enroll(cohort_id, student_id, now) A->>R: get(cohort_id) R-->>A: поток A->>C: enroll(student_id, now) C-->>A: номер записи A->>R: save(поток) A->>C: pull_events() C-->>A: [StudentEnrolled] A->>B: publish(события) B-->>B: Оплата выставляет счёт, Уведомления пишут письмо A-->>W: события
Одна транзакция — один агрегат
Заголовок раздела «Одна транзакция — один агрегат»В одной транзакции прикладной сервис меняет один агрегат. Это правило Вона Вернона («Effective Aggregate Design»): если команде нужно изменить и другие агрегаты, они догоняют через доменные события — согласованность в конечном счётесогласованность в конечном счётеEN eventual consistencyДанные в разных местах (агрегат и модель чтения, два контекста) какое-то время расходятся, но без новых изменений сходятся к одному значению. Плата за независимость частей системы; бизнес обычно и так живёт с задержками и исправлениями. (eventual consistency). У Эванса: правило, охватывающее несколько агрегатов, не обязано выполняться в каждый момент.
Платёж в Оплате и поток в Наборе — разные агрегаты и даже контексты: Оплата фиксирует PaymentReceived, а Набор в
своей транзакции подтверждает запись. Короткий миг между ними бизнесу не мешает. Если эксперты требуют
атомарности, проверьте границу агрегата (урок DDD.3.2).
Бизнес-правило в прикладном сервисе — запах
Заголовок раздела «Бизнес-правило в прикладном сервисе — запах»def enroll(self, cohort_id, student_id, now): cohort = self._cohorts.get(cohort_id) if cohort.free_seats <= 0: # правило вместимости — в сервисе! raise NoFreeSeats(...) if now >= cohort.starts_at: # и правило закрытия записи raise EnrollmentClosed(...) cohort.enrollments_list.append(Enrollment(...)) # агрегат — просто список self._cohorts.save(cohort)В сценарии записи через поддержку правило появится копией, в импорте его забудут, а агрегат больше не защищает себя — прямая дорога к анемичной модели из урока DDD.3.4.
| Доменный сервис | Прикладной сервис | |
|---|---|---|
| Что внутри | бизнес-правило: цена со скидками, перевод между счетами | порядок шагов сценария |
| Знает о хранилище и шине | нет — всё приходит аргументами | да — через порты |
| Состояние | нет | нет, только ссылки на порты |
| Слой | домен | приложение |
| Пример | enrollment_price(cohort, now, is_alumnus) | EnrollmentService.enroll(cohort_id, student_id, now) |
| Как тестировать | таблица примеров, без фейков | на фейках: хранилище и шина в памяти |
Различить помогает вопрос: узнает ли эксперт в этом коде своё правило? В enrollment_price — да; в
EnrollmentService.enroll — нет, и это нормально.
Практика
Заголовок раздела «Практика»Полный сценарий «записаться на поток»
Заголовок раздела «Полный сценарий «записаться на поток»»Соберём всё вместе. Структура проекта — плоский пакет, как в домашней работе модуля:
school/├── enrollment/│ ├── __init__.py # пустой: папка — пакет Python│ ├── money.py│ ├── events.py # доменные события│ ├── model.py # агрегат Cohort и фабрика│ ├── pricing.py # доменный сервис (из раздела выше)│ └── service.py # прикладной сервис├── demo_enroll.py└── test_events.pyЕсли вы решили задание урока DDD.3.1, возьмите свой money.py: он строже и умеет больше. Сокращённая версия:
from dataclasses import dataclassfrom decimal import ROUND_HALF_UP, Decimal
@dataclass(frozen=True)class Money: """Деньги: копейки и валюта. Сокращённая версия из урока DDD.3.1 — без проверок и сложения.""" amount: int currency: str
def __mul__(self, factor): kopecks = (Decimal(self.amount) * factor).quantize(Decimal(1), rounding=ROUND_HALF_UP) return Money(int(kopecks), self.currency)
def __str__(self): rubles = f"{self.amount // 100:,}".replace(",", " ") return f"{rubles},{self.amount % 100:02d} {self.currency}""""Доменные события контекста «Набор»: неизменяемые факты в прошедшем времени."""from dataclasses import dataclassfrom datetime import datetimefrom enum import Enum
from .money import Money
class CancellationReason(Enum): """Причина отмены — решение экспертов на Event Storming (урок DDD.2.3).""" BY_STUDENT = "отменена студентом" UNPAID = "не оплачена за 48 часов"
@dataclass(frozen=True)class DomainEvent: """Базовый класс доменных событий."""
@dataclass(frozen=True)class StudentEnrolled(DomainEvent): """СтудентЗаписан: студент занял место на потоке, запись ожидает оплаты.""" cohort_id: str enrollment_id: str student_id: str price: Money # на какую сумму Оплате выставить счёт expires_at: datetime # когда неоплаченная запись отменится (48 часов) occurred_at: datetime
@dataclass(frozen=True)class EnrollmentConfirmed(DomainEvent): """ЗаписьПодтверждена: платёж по записи получен.""" cohort_id: str enrollment_id: str student_id: str occurred_at: datetime
@dataclass(frozen=True)class EnrollmentCancelled(DomainEvent): """ЗаписьОтменена — одно событие для обеих причин; сумму возврата агрегат решил при отмене.""" cohort_id: str enrollment_id: str student_id: str reason: CancellationReason refund: Money occurred_at: datetime
@dataclass(frozen=True)class SeatReleased(DomainEvent): """МестоОсвободилось: на потоке стало на одно свободное место больше.""" cohort_id: str occurred_at: datetimeМодель — агрегат из урока DDD.3.2 без листа ожидания, но с ценой, версией, событиями и фабрикой; cancel
возвращает сумму возврата.
"""Агрегат «Поток» из урока DDD.3.2 (без листа ожидания) — с ценой, событиями и фабрикой."""from dataclasses import dataclass, replacefrom datetime import datetime, timedeltafrom enum import Enum
from .events import (CancellationReason, EnrollmentCancelled, EnrollmentConfirmed, SeatReleased, StudentEnrolled)from .money import Money
PAYMENT_PERIOD = timedelta(hours=48) # условия учебного примера: 48 часов на оплату,FULL_REFUND_PERIOD = timedelta(days=7) # 7 × 24 часа до старта — полный возврат
class EnrollmentStatus(Enum): PENDING_PAYMENT = "ожидает оплаты" CONFIRMED = "подтверждена" CANCELLED = "отменена"
class DomainError(Exception): """Нарушено правило предметной области."""
class NoFreeSeats(DomainError): """На потоке нет свободных мест."""
class AlreadyEnrolled(DomainError): """Студент уже записан на поток."""
class EnrollmentClosed(DomainError): """Запись на поток закрыта: поток уже стартовал."""
class EnrollmentNotFound(DomainError): """В потоке нет записи с таким номером."""
class InvalidStatusTransition(DomainError): """Такой переход статуса запрещён, например повторная отмена."""
@dataclass(frozen=True)class Course: """Курс глазами Набора: только то, что нужно, чтобы открыть поток.""" slug: str price: Money
@dataclassclass Enrollment: """Запись на поток — сущность внутри агрегата. Меняет её только корень.""" enrollment_id: str student_id: str enrolled_at: datetime status: EnrollmentStatus = EnrollmentStatus.PENDING_PAYMENT
@property def takes_seat(self): return self.status != EnrollmentStatus.CANCELLED
def __eq__(self, other): # сущность: равенство по номеру (урок DDD.3.1) return isinstance(other, Enrollment) and other.enrollment_id == self.enrollment_id
def __hash__(self): return hash(self.enrollment_id)
class Cohort: """Поток курса — корень агрегата. Состояние меняется только его методами."""
def __init__(self, cohort_id, starts_at, capacity, price, enrollments=(), version=0): # Конструктор принимает готовое состояние — так поток восстанавливают из хранилища. if isinstance(capacity, bool) or not isinstance(capacity, int) or capacity < 1: raise ValueError(f"Вместимость потока — целое число от 1, а получено {capacity!r}") self.cohort_id = cohort_id self.starts_at = starts_at self.capacity = capacity self.price = price self.version = version # для оптимистической блокировки (уроки DDD.3.2 и DDD.3.4) self._enrollments = {e.enrollment_id: replace(e) for e in enrollments} self._next_number = len(self._enrollments) + 1 self._events = []
@classmethod def open(cls, course, starts_at, capacity, now): """Фабрика: открыть новый поток курса. Здесь — правила рождения и выдача номера.""" if starts_at <= now: raise ValueError(f"Нельзя открыть поток со стартом в прошлом: {starts_at:%d.%m.%Y %H:%M}") cohort_id = f"{course.slug}-{starts_at:%Y-%m}" # номер по правилам школы: курс-год-месяц return cls(cohort_id, starts_at, capacity, price=course.price)
# ---------- чтение: наружу только копии ---------- @property def free_seats(self): return self.capacity - sum(e.takes_seat for e in self._enrollments.values())
@property def enrollments(self): return tuple(replace(e) for e in self._enrollments.values())
# ---------- команды ---------- def enroll(self, student_id, now): """Записать студента. Заодно это фабричный метод сущности Enrollment.""" if now >= self.starts_at: raise EnrollmentClosed(f"Запись на поток {self.cohort_id} закрыта: поток стартовал") if any(e.student_id == student_id and e.takes_seat for e in self._enrollments.values()): raise AlreadyEnrolled(f"Студент {student_id} уже записан на поток {self.cohort_id}") if self.free_seats <= 0: raise NoFreeSeats(f"На потоке {self.cohort_id} нет свободных мест") enrollment_id = f"{self.cohort_id}-{self._next_number}" self._next_number += 1 self._enrollments[enrollment_id] = Enrollment(enrollment_id, student_id, enrolled_at=now) self._events.append(StudentEnrolled( self.cohort_id, enrollment_id, student_id, price=self.price, expires_at=now + PAYMENT_PERIOD, occurred_at=now)) return enrollment_id
def confirm(self, enrollment_id, now): """Подтвердить запись: платёж по ней получен.""" enrollment = self._find(enrollment_id) if enrollment.status != EnrollmentStatus.PENDING_PAYMENT: raise InvalidStatusTransition(f"Запись {enrollment_id} не ожидает оплаты") enrollment.status = EnrollmentStatus.CONFIRMED self._events.append(EnrollmentConfirmed( self.cohort_id, enrollment_id, enrollment.student_id, occurred_at=now))
def cancel(self, enrollment_id, now, reason=CancellationReason.BY_STUDENT): """Отменить запись. Возврат — только при отмене студентом оплаченной записи за 7 суток и более.""" enrollment = self._find(enrollment_id) if enrollment.status == EnrollmentStatus.CANCELLED: raise InvalidStatusTransition(f"Запись {enrollment_id} уже отменена") overdue = (enrollment.status == EnrollmentStatus.PENDING_PAYMENT and now - enrollment.enrolled_at >= PAYMENT_PERIOD) if reason == CancellationReason.UNPAID and not overdue: # причине не верим на слово (DDD.3.2) raise InvalidStatusTransition(f"Запись {enrollment_id} не ждёт оплаты 48 часов") paid = enrollment.status == EnrollmentStatus.CONFIRMED if reason == CancellationReason.BY_STUDENT and paid and self.starts_at - now >= FULL_REFUND_PERIOD: refund = self.price else: refund = Money(0, self.price.currency) # не платил или отменил поздно enrollment.status = EnrollmentStatus.CANCELLED self._events.append(EnrollmentCancelled( self.cohort_id, enrollment_id, enrollment.student_id, reason, refund=refund, occurred_at=now)) self._events.append(SeatReleased(self.cohort_id, occurred_at=now)) return refund
def pull_events(self): """Отдать накопленные события и очистить список: каждое событие уходит один раз.""" events, self._events = self._events, [] return events
def _find(self, enrollment_id): if enrollment_id not in self._enrollments: raise EnrollmentNotFound(f"В потоке {self.cohort_id} нет записи {enrollment_id}") return self._enrollments[enrollment_id]Прикладной сервис: общие шаги трёх сценариев вынесены в _execute, а команда — функция от потока, записанная
через lambda. Время now берёт из часов вызывающий код (веб-обработчик, плановая задача), а тест подставляет
нужный момент. В домашней работе сервис такой же, только без публикатора: события он возвращает тестам.
"""Прикладной сервис контекста «Набор»: сценарии использования — без бизнес-правил."""from .events import CancellationReason
class EnrollmentService: def __init__(self, cohorts, publisher): self._cohorts = cohorts # хранилище потоков: get и save (урок DDD.3.4) self._publisher = publisher # передаёт события подписчикам: publish(events)
def enroll(self, cohort_id, student_id, now): """Сценарий «записаться на поток».""" return self._execute(cohort_id, lambda cohort: cohort.enroll(student_id, now))
def confirm(self, cohort_id, enrollment_id, now): """Сценарий «подтвердить запись» — Набор запускает его, узнав о платеже.""" return self._execute(cohort_id, lambda cohort: cohort.confirm(enrollment_id, now))
def cancel(self, cohort_id, enrollment_id, now, reason=CancellationReason.BY_STUDENT): """Сценарий «отменить запись» — по просьбе студента или по таймеру 48 часов.""" return self._execute(cohort_id, lambda cohort: cohort.cancel(enrollment_id, now, reason))
def _execute(self, cohort_id, command): """Общие шаги любого сценария над одним потоком.""" cohort = self._cohorts.get(cohort_id) # 1. загрузить агрегат command(cohort) # 2. команда агрегату: правила — там self._cohorts.save(cohort) # 3. сохранить: одна транзакция — один агрегат events = cohort.pull_events() # 4. забрать события self._publisher.publish(events) # 5. опубликовать — после сохранения return eventsОсталось собрать приложение: временное хранилище-словарь (настоящий репозиторий — в уроке DDD.3.4), шину событий в памяти и подписчиков из других контекстов, о которых агрегат ничего не знает.
from collections import defaultdictfrom datetime import datetime
from enrollment.events import CancellationReason, EnrollmentCancelled, SeatReleased, StudentEnrolledfrom enrollment.model import Cohort, Course, NoFreeSeatsfrom enrollment.money import Moneyfrom enrollment.service import EnrollmentService
class DictCohorts: # временное хранилище потоков в словаре def __init__(self, *cohorts): self._items = {c.cohort_id: c for c in cohorts}
def get(self, cohort_id): return self._items[cohort_id]
def save(self, cohort): self._items[cohort.cohort_id] = cohort
class InMemoryEventBus: # шина событий в памяти: тип события → обработчики def __init__(self): self._handlers = defaultdict(list)
def subscribe(self, event_type, handler): self._handlers[event_type].append(handler)
def publish(self, events): for event in events: for handler in self._handlers[type(event)]: handler(event)
# --- сборка приложения ---------------------------------------------------------python = Course("python", price=Money(3_000_000, "RUB")) # 30 000 руб. в копейкахcohort = Cohort.open(python, datetime(2026, 11, 2, 10, 0), capacity=2, now=datetime(2026, 9, 1))bus = InMemoryEventBus()service = EnrollmentService(DictCohorts(cohort), bus)
bus.subscribe(StudentEnrolled, lambda e: print( f" [Оплата] выставить счёт по записи {e.enrollment_id} на {e.price} до {e.expires_at:%d.%m %H:%M}"))bus.subscribe(StudentEnrolled, lambda e: print( f" [Уведомления] письмо {e.student_id}: вы записаны на поток {e.cohort_id}"))bus.subscribe(EnrollmentCancelled, lambda e: print( f" [Оплата] запись {e.enrollment_id} {e.reason.value}, вернуть {e.refund}"))bus.subscribe(SeatReleased, lambda e: print( f" [Набор] на потоке {e.cohort_id} освободилось место — предложить первому в листе ожидания"))
# --- сценарии ------------------------------------------------------------------print("1 октября, 10:00 — Анна записывается:")events = service.enroll("python-2026-11", "anna", datetime(2026, 10, 1, 10, 0))print(" события:", [type(e).__name__ for e in events])
print("1 октября, 11:00 — Борис записывается:")service.enroll("python-2026-11", "boris", datetime(2026, 10, 1, 11, 0))
print("1 октября, 12:00 — Вера записывается:")try: service.enroll("python-2026-11", "vera", datetime(2026, 10, 1, 12, 0))except NoFreeSeats as error: print(" отказ:", error, "— событий нет")
print("1 октября, 13:00 — платёж Анны получен:")events = service.confirm("python-2026-11", "python-2026-11-1", datetime(2026, 10, 1, 13, 0))print(" события:", [type(e).__name__ for e in events])
print("3 октября, 11:00 — Борис не оплатил за 48 часов, срабатывает таймер:")service.cancel("python-2026-11", "python-2026-11-2", datetime(2026, 10, 3, 11, 0), CancellationReason.UNPAID)
print("20 октября, 10:00 — Анна отменяет запись, до старта 13 суток:")events = service.cancel("python-2026-11", "python-2026-11-1", datetime(2026, 10, 20, 10, 0))print(" события:", [type(e).__name__ for e in events])1 октября, 10:00 — Анна записывается: [Оплата] выставить счёт по записи python-2026-11-1 на 30 000,00 RUB до 03.10 10:00 [Уведомления] письмо anna: вы записаны на поток python-2026-11 события: ['StudentEnrolled']1 октября, 11:00 — Борис записывается: [Оплата] выставить счёт по записи python-2026-11-2 на 30 000,00 RUB до 03.10 11:00 [Уведомления] письмо boris: вы записаны на поток python-2026-111 октября, 12:00 — Вера записывается: отказ: На потоке python-2026-11 нет свободных мест — событий нет1 октября, 13:00 — платёж Анны получен: события: ['EnrollmentConfirmed']3 октября, 11:00 — Борис не оплатил за 48 часов, срабатывает таймер: [Оплата] запись python-2026-11-2 не оплачена за 48 часов, вернуть 0,00 RUB [Набор] на потоке python-2026-11 освободилось место — предложить первому в листе ожидания20 октября, 10:00 — Анна отменяет запись, до старта 13 суток: [Оплата] запись python-2026-11-1 отменена студентом, вернуть 30 000,00 RUB [Набор] на потоке python-2026-11 освободилось место — предложить первому в листе ожидания события: ['EnrollmentCancelled', 'SeatReleased']Что видно в выводе: команда даёт ноль, одно или два события; подписчики реагируют независимо, и новая реакция —
просто ещё один подписчик без правки модели; отказ событий не породил; две отмены — одно событие с разными
причинами, а возврат посчитал агрегат. В настоящей школе confirm вызовет подписчик на PaymentReceived, а отмену
по таймеру — плановая задача (в домашней работе — метод expire_unpaid).
Тесты событий
Заголовок раздела «Тесты событий»События — выход агрегата, и тестировать их удобно как выход функции: дано — когда — тогда такие события. Frozen dataclass сравнивается по полям, поэтому событие можно сравнить целиком.
from datetime import datetime
from enrollment.events import CancellationReason, EnrollmentCancelled, SeatReleased, StudentEnrolledfrom enrollment.model import Cohort, Coursefrom enrollment.money import Money
NOW = datetime(2026, 10, 1, 10, 0)
def open_cohort(): course = Course("python", price=Money(3_000_000, "RUB")) return Cohort.open(course, starts_at=datetime(2026, 11, 2, 10, 0), capacity=2, now=NOW)
def test_enrollment_records_student_enrolled_with_payment_deadline(): """СтудентЗаписан хранит срок оплаты: 48 часов от момента записи.""" cohort = open_cohort() enrollment_id = cohort.enroll("anna", NOW) assert cohort.pull_events() == [StudentEnrolled( "python-2026-11", enrollment_id, "anna", price=Money(3_000_000, "RUB"), expires_at=datetime(2026, 10, 3, 10, 0), occurred_at=NOW)]
def test_cancellation_records_two_facts_with_reason(): """Одна команда — два факта: ЗаписьОтменена (с причиной) и МестоОсвободилось.""" cohort = open_cohort() enrollment_id = cohort.enroll("anna", NOW) cohort.pull_events() cohort.cancel(enrollment_id, datetime(2026, 10, 3, 10, 0), CancellationReason.UNPAID) # ровно 48 часов cancelled, released = cohort.pull_events() assert isinstance(cancelled, EnrollmentCancelled) and isinstance(released, SeatReleased) assert cancelled.reason == CancellationReason.UNPAID and cancelled.refund == Money(0, "RUB")
def test_events_are_given_out_once(): """pull_events() очищает список: второй вызов ничего не возвращает.""" cohort = open_cohort() cohort.enroll("anna", NOW) assert len(cohort.pull_events()) == 1 assert cohort.pull_events() == []$ python -m pytest -q test_events.py... [100%]3 passed in 0.01sЗапускайте python -m pytest из папки school/: так Python найдёт пакет enrollment.
Упражнение
Заголовок раздела «Упражнение»Найдите четыре проблемы (15 минут). Новичок написал сценарий «перенести запись на другой поток»:
def transfer(self, from_cohort_id, to_cohort_id, enrollment_id, student_id): source = self._cohorts.get(from_cohort_id) target = self._cohorts.get(to_cohort_id) if target.starts_at <= source.starts_at: raise ValueError("Переносить можно только на более поздний поток") now = datetime.now() source.cancel(enrollment_id, now) target.enroll(student_id, now) self._publisher.publish(source.pull_events() + target.pull_events()) self._cohorts.save(source) self._cohorts.save(target)Разбор
- Бизнес-правило в прикладном сервисе: «только на более поздний поток» — место ему в модели.
datetime.now()вместо времени снаружи: граничные моменты не проверить тестом.- Публикация до сохранения: упади
save(target)— деньги уже вернули, письма разослали. - Два агрегата в одной операции: первый
saveпрошёл, второй нет — студент без обоих потоков.
И пятая, предметная: перенос — не «отмена плюс запись». По EnrollmentCancelled Оплата вернёт деньги, а при
переносе они должны остаться в школе. У переноса свой язык и свои события; эту фичу вы спроектируете в домашней
работе программы SDD (sdd-hw01-spec).
Проверь себя
Заголовок раздела «Проверь себя»- Доменное событие — неизменяемый факт в прошедшем времени на едином языке: идентификаторы, момент и факты для подписчиков (цена, срок оплаты, сумма возврата), но не сам агрегат. Отказ — исключение, а не событие.
- Агрегат копит события, прикладной сервис забирает их через
pull_events()и публикует после сохранения. - Доменный сервис — правило без состояния, которому не место ни в одной сущности (цена со скидками, перевод между счетами). Сначала пробуйте агрегат или значение: лишние сервисы ведут к анемичной модели.
- Фабрика держит правила рождения и выдаёт номера: фабричный метод корня, отдельная фабрика
Cohort.openи восстановление из хранилища, которое правил рождения не проверяет и событий не порождает. - Прикладной сервис ведёт сценарий — загрузить, вызвать, сохранить, опубликовать — без бизнес-правил и меняет в одной транзакции один агрегат; остальные догоняют через события.
Дальше: урок DDD.3.4 — репозитории и архитектура. Заменим словарь настоящим репозиторием с оптимистической блокировкой, разберём порты и адаптеры, гексагональную и чистую архитектуру, научимся тестировать домен без базы и посмотрим, чего избегать: анемичной модели, сервисов-всезнаек и утечек ORM в домен.