Репозитории и архитектура: модель в центре
Как хранить агрегаты, не пуская базу данных в модель: репозиторий как иллюзия коллекции, порт рядом с моделью и адаптеры в памяти и на SQLite с оптимистической блокировкой; слоистая, гексагональная и чистая архитектура через направление зависимостей; тесты без базы, структура Python-проекта по контекстам и слоям и три болезни — анемичная модель, сервисы-всезнайки и утечки ORM.
Зачем это нужно
Заголовок раздела «Зачем это нужно»В уроке DDD.3.3 прикладной сервис хранил потоки в словаре DictCohorts. Для
демонстрации хватало, но у словаря три недостатка. Он отдаёт тот же объект, что лежит внутри, поэтому сервис, который
забыл вызвать save, всё равно «работает» — до первого запуска на настоящей базе. Он не замечает параллельных
изменений, а в уроке DDD.3.2 вы видели, чем это кончается:
потерянным обновлениемпотерянное обновлениеEN lost updateОшибка параллельной работы: два запроса прочитали одну версию данных, оба изменили и сохранили — второй молча затёр изменения первого. Ловится оптимистической блокировкой по номеру версии. (lost update). И он ничего не говорит о главном вопросе: где в
проекте живёт SQL и кто от кого зависит.
Обычный путь без DDD — модель наследует класс ORM, правила пишутся вперемешку с запросами, а тест записи на поток поднимает базу данных. Через год тесты идут двадцать минут, и их перестают запускать. В этом уроке мы сделаем наоборот: модель в центре и ничего не знает о хранении, база подключается снаружи через порт, а сотни тестов правил идут за доли секунды. Заодно разберём архитектуры, которые возводят этот приём в принцип, и болезни, которые появляются, когда о нём забывают.
Репозиторий
Заголовок раздела «Репозиторий»Иллюзия коллекции
Заголовок раздела «Иллюзия коллекции»Эванс описывает репозиторийрепозиторий (шаблон DDD)EN repositoryИллюзия коллекции агрегатов в памяти: add, get, save — и никакого SQL снаружи. Один репозиторий на тип агрегата (не на таблицу); интерфейс-порт объявлен рядом с моделью, реализация-адаптер — в инфраструктуре. Не путать с git-репозиторием. (repository) так: для каждого типа агрегатов, к которым нужен доступ, заведите объект, который создаёт иллюзию коллекции всех таких объектов в памяти. Авторы «Cosmic Python» предлагают мысленный эксперимент: будь у нас бесконечная память, базы данных не понадобились бы — мы бы положили потоки в коллекцию, доставали нужный и меняли его. Репозиторий даёт модели и прикладному сервису именно такой интерфейс, а как коллекция устроена на самом деле — в памяти, в SQLite, в PostgreSQL, — решает реализация.
Отсюда скромный набор методов: add — положить новый поток, get — достать по номеру, save — сохранить
изменения. Ни execute(sql), ни update_capacity(...): менять поток можно только его методами, а репозиторий лишь
достаёт и кладёт. Если экрану нужен хитрый отчёт — «свободные места по всем потокам курса», — это работа модели
чтения (урок DDD.4.1), а не нового метода репозитория.
Один репозиторий на агрегат
Заголовок раздела «Один репозиторий на агрегат»Репозиторий заводят на тип агрегата, а не на таблицу. У потока есть записи, но репозитория записей нет: запись
меняется только через корень, иначе инварианты из урока DDD.3.2 обойдёт первый же enrollments_repo.save(...).
Руководство Microsoft по микросервисам формулирует жёстко: репозиторий — единственный канал, которым меняют данные,
и у него связь один к одному с корнем агрегата. Читать данные в обход (для отчётов и экранов) можно, менять — нет.
Порт рядом с моделью
Заголовок раздела «Порт рядом с моделью»Интерфейс репозитория говорит словами модели, поэтому объявляют его рядом с моделью, а реализуют в инфраструктуре.
В Python для этого есть абстрактные базовые классы из модуля abc: класс с методами @abstractmethod нельзя создать,
пока наследник не реализует их все. Такой абстрактный класс и есть портпорты и адаптерыEN ports and adaptersПорт — интерфейс на границе приложения, выраженный его словами (CohortRepository, Clock); адаптер — реализация порта для конкретной технологии (SQLite, память, HTTP). Ведущие (driving) адаптеры вызывают приложение, ведомые (driven) вызываются им.; классы,
которые его реализуют, — адаптеры.
"""Порт «Потоки школы» и его адаптер в памяти."""import copyfrom abc import ABC, abstractmethod
class CohortNotFound(Exception): """В хранилище нет потока с таким номером."""
class ConcurrencyError(Exception): """Поток изменили параллельно: версия в хранилище новее той, что была загружена."""
class CohortRepository(ABC): """Порт: все потоки школы — как будто это коллекция в памяти."""
@abstractmethod def add(self, cohort): """Положить новый поток."""
@abstractmethod def get(self, cohort_id): """Достать поток целиком или бросить CohortNotFound."""
@abstractmethod def save(self, cohort): """Сохранить изменения или бросить ConcurrencyError, если поток успели изменить."""
class InMemoryCohortRepository(CohortRepository): """Адаптер для тестов и прототипов: хранит и отдаёт копии, как настоящая база."""
def __init__(self): self._items = {} # номер потока → сохранённая копия
def add(self, cohort): if cohort.cohort_id in self._items: raise ValueError(f"Поток {cohort.cohort_id} уже есть") self._items[cohort.cohort_id] = self._snapshot(cohort)
def get(self, cohort_id): if cohort_id not in self._items: raise CohortNotFound(f"Потока {cohort_id} нет") return copy.deepcopy(self._items[cohort_id]) # каждому запросу — своя копия
def save(self, cohort): stored = self._items.get(cohort.cohort_id) if stored is None: raise CohortNotFound(f"Потока {cohort.cohort_id} нет — сначала add()") if stored.version != cohort.version: raise ConcurrencyError(f"Поток {cohort.cohort_id} изменили параллельно: " f"загружена версия {cohort.version}, в хранилище {stored.version}") cohort.version += 1 self._items[cohort.cohort_id] = self._snapshot(cohort)
@staticmethod def _snapshot(cohort): snapshot = copy.deepcopy(cohort) snapshot.pull_events() # неопубликованные события в хранилище не попадают — как и в базу return snapshotТри детали делают этот фейкфейк (рабочая подделка для тестов)EN fakeТестовый двойник с настоящим, но упрощённым поведением: репозиторий в словаре, часы с фиксированным временем. В отличие от мока не проверяет, какие методы вызвали, а просто работает. Чтобы фейк не врал, его прогоняют теми же контрактными тестами, что и настоящий адаптер. честным, то есть похожим на настоящую базу:
- копии на входе и выходе.
copy.deepcopyсоздаёт полную независимую копию объекта со всеми вложенными. Два запроса получают два разных потока, а изменения безsaveтеряются — ровно как в базе; - проверка версии — та же оптимистическая блокировкаоптимистическая блокировкаEN optimistic locking, optimistic concurrency controlСпособ поймать параллельные изменения без блокировки на время работы: у агрегата есть номер версии, при сохранении хранилище сверяет его с текущим. Версия выросла — кто-то успел раньше, сохранение отклоняется (ConcurrencyError), команду повторяют на свежих данных., что в уроке DDD.3.2: загружена версия 4, а в хранилище уже 5 — значит, кто-то успел раньше, и сохранение отклоняется;
- события не сохраняются. Прикладной сервис забирает события через
pull_events()послеsave, а копия в хранилище должна быть без них — иначе следующийgetвернул бы старые события ещё раз.
Наш репозиторий требует явного save. Вон Вернон называет такие репозитории ориентированными на хранение
(persistence-oriented), в отличие от ориентированных на коллекцию, где достаточно изменить объект, а изменения
отследит ORM. В Python без ORM явный save проще и честнее: видно, где проверяется версия.
Адаптер на SQLite
Заголовок раздела «Адаптер на SQLite»Порт позволяет подключить настоящую базу, не трогая модель. SQLite входит в стандартную библиотеку Python
(модуль sqlite3), и база может жить прямо в памяти — sqlite3.connect(":memory:"). Поток храним одной строкой:
номер, версия и состояние в JSON.
"""Адаптер «Потоки в SQLite»: тот же порт, что у InMemoryCohortRepository, но с настоящей базой."""import jsonfrom datetime import datetime
from .model import Cohort, Enrollment, EnrollmentStatusfrom .money import Moneyfrom .repository import CohortNotFound, CohortRepository, ConcurrencyError
class SqliteCohortRepository(CohortRepository): def __init__(self, connection): self._db = connection self._db.execute("CREATE TABLE IF NOT EXISTS cohorts " "(cohort_id TEXT PRIMARY KEY, version INTEGER NOT NULL, data TEXT NOT NULL)")
def add(self, cohort): self._db.execute("INSERT INTO cohorts VALUES (?, ?, ?)", (cohort.cohort_id, cohort.version, _to_json(cohort))) self._db.commit()
def get(self, cohort_id): row = self._db.execute("SELECT version, data FROM cohorts WHERE cohort_id = ?", (cohort_id,)).fetchone() if row is None: raise CohortNotFound(f"Потока {cohort_id} нет") return _from_json(cohort_id, row[1], version=row[0])
def save(self, cohort): # Оптимистическая блокировка одной командой: обновить, только если версия не изменилась updated = self._db.execute( "UPDATE cohorts SET data = ?, version = version + 1 WHERE cohort_id = ? AND version = ?", (_to_json(cohort), cohort.cohort_id, cohort.version)).rowcount self._db.commit() if updated == 0: raise ConcurrencyError(f"Поток {cohort.cohort_id} изменили параллельно " f"(загружена версия {cohort.version}) или его нет") cohort.version += 1
# Отображение «агрегат ↔ строка таблицы» живёт здесь, в адаптере: модель о JSON и SQL не знает.def _to_json(cohort): return json.dumps({ "starts_at": cohort.starts_at.isoformat(), "capacity": cohort.capacity, "price": [cohort.price.amount, cohort.price.currency], "enrollments": [[e.enrollment_id, e.student_id, e.enrolled_at.isoformat(), e.status.name] for e in cohort.enrollments], })
def _from_json(cohort_id, text, version): """Восстановление (урок DDD.3.3): конструктор Cohort, без правил рождения и без событий.""" data = json.loads(text) enrollments = [Enrollment(number, student, datetime.fromisoformat(at), EnrollmentStatus[status]) for number, student, at, status in data["enrollments"]] return Cohort(cohort_id, datetime.fromisoformat(data["starts_at"]), data["capacity"], Money(*data["price"]), enrollments, version)Главная строка — UPDATE ... WHERE cohort_id = ? AND version = ?. База обновит строку, только если версия всё ещё
та, что мы загрузили, и сообщит в rowcount, сколько строк изменила. Ноль строк — нас опередили. Проверка и запись
идут одной командой, поэтому между ними никто не вклинится. Отображение объекта в строку и обратно лежит в адаптере, а
восстановление идёт через конструктор Cohort — ровно как договорились в уроке DDD.3.3. Модель не импортирует ни
json, ни sqlite3 и даже не знает, что её хранят: это свойство называют
независимостью модели от хранениянезависимость модели от храненияEN persistence ignoranceПринцип: классы модели не знают, как и где их хранят, — не наследуют классы ORM, не импортируют драйвер базы. Отображение «объект ↔ строка» живёт в адаптере. (persistence ignorance).
В адаптере есть упрощение: commit вызывается внутри save. Если сценарию нужно зафиксировать вместе несколько
изменений — агрегат и его события в таблице outbox из урока DDD.4.2, — транзакцией управляет
единица работыединица работыEN unit of workШаблон из книги Мартина Фаулера «Patterns of Enterprise Application Architecture» (2002): отслеживает объекты, изменённые в одной бизнес-операции, и фиксирует их одной транзакцией — всё или ничего. В Python удобно оформляется контекстным менеджером; сессия SQLAlchemy уже реализует этот шаблон. (unit of work) из книги Мартина Фаулера «Patterns of Enterprise
Application Architecture» (2002). В Python её удобно оформить контекстным менеджером: with uow: открывает
транзакцию, uow.commit() фиксирует всё сразу, а выход из блока без commit откатывает. Так она устроена в главе 6
«Cosmic Python»; сессия SQLAlchemy и transaction.atomic в Django — готовые единицы работы.
Повтор при конфликте
Заголовок раздела «Повтор при конфликте»Что делать с ConcurrencyError? Обычный ответ, который дают и «Cosmic Python», и урок DDD.3.2, — повторить команду с
начала на свежих данных. Это работа прикладного сервиса: в service.py из урока DDD.3.3 добавьте импорт и замените
_execute:
# в начало service.py:from .repository import ConcurrencyError
MAX_ATTEMPTS = 3
# в класс EnrollmentService вместо прежнего _execute: def _execute(self, cohort_id, command): """Загрузить → команда → сохранить → опубликовать. Проиграли гонку — повторить с начала.""" for attempt in range(1, MAX_ATTEMPTS + 1): cohort = self._cohorts.get(cohort_id) # каждый раз — свежая версия потока command(cohort) # доменные исключения летят дальше как есть try: self._cohorts.save(cohort) except ConcurrencyError: if attempt == MAX_ATTEMPTS: raise # конфликтов слишком много — сдаёмся честно continue # кто-то успел раньше: повторить на свежих данных events = cohort.pull_events() self._publisher.publish(events) return eventsПри повторе правила проверяются заново: если параллельный запрос занял последнее место, вторая попытка честно
получит NoFreeSeats. События проигравшей попытки пропадают вместе с её копией потока, и подписчики о ней не узнают.
Оптимистическая блокировка не единственная. Пессимистическая (SELECT ... FOR UPDATE) запирает строку на время
сценария: конфликтов нет, но остальные ждут и возможны взаимные блокировки. Оптимистическая хороша, когда конфликты
редки. В пик открытия продаж поток правят сотни команд за минуты — канва агрегата из урока DDD.3.2 (раздел
«Пропускная способность») уже оценила шанс конфликта как высокий. Тогда повторы станут частыми, и это довод ещё раз
проверить границу агрегата.
Архитектура: модель в центре
Заголовок раздела «Архитектура: модель в центре»Слоистая архитектура и её беды
Заголовок раздела «Слоистая архитектура и её беды»Слоистая архитектураслоистая архитектураEN layered architectureКод делится на слои с правилом «вызывать можно только нижние»: интерфейс, приложение, домен, инфраструктура (так у Эванса). Беда классического варианта — домен зависит от слоя базы данных, лежащего под ним. (layered architecture) делит код на слои, и каждый слой обращается только к нижним. У Эванса слоёв четыре: интерфейс пользователя, приложение, предметная область, инфраструктура. Это лучше «кома грязи»: у кода появляются места. Но в классическом варианте база данных лежит в самом низу, и модель зависит от неё. Отсюда беды:
- модель импортирует ORM и драйвер базы, и тест правила «не больше мест, чем вместимость» поднимает базу;
- проектирование начинается с таблиц, а не с языка экспертов, — модель повторяет схему базы;
- правила утекают вверх, в обработчики запросов, и вниз, в хранимые процедуры и триггеры;
- сменить хранилище или добавить второе (кэш, поиск) — значит править предметную область.
Инверсия зависимостей
Заголовок раздела «Инверсия зависимостей»Лечение — принцип инверсии зависимостейпринцип инверсии зависимостейEN dependency inversion principle (DIP)Буква D в SOLID Роберта Мартина: модули верхнего уровня (модель) не зависят от модулей нижнего уровня (база, почта) — оба зависят от абстракций, и абстракции принадлежат верхнему уровню. Основа гексагональной и чистой архитектуры. (dependency inversion
principle), буква D в SOLID Роберта Мартина: модули верхнего уровня (модель, сценарии) не должны зависеть от модулей
нижнего уровня (база, почта); и те и другие зависят от абстракций. Ключевое — абстракция принадлежит верхнему уровню.
CohortRepository объявлен словами модели рядом с ней, а SqliteCohortRepository импортирует модель и реализует
её порт. Стрелка зависимости между моделью и базой развернулась: теперь база зависит от модели.
Гексагональная архитектура: порты и адаптеры
Заголовок раздела «Гексагональная архитектура: порты и адаптеры»Алистер Кокберн в 2005 году описал гексагональную архитектуругексагональная архитектураEN hexagonal architectureАрхитектура Алистера Кокберна (2005), второе название — «порты и адаптеры»: приложение в центре, внешний мир (пользователи, тесты, база, почта) подключается через порты и адаптеры. Шестиугольник — просто удобная картинка, число сторон не важно. (hexagonal architecture), второе её имя — «порты и адаптеры». Цель он сформулировал так: приложение должно одинаково управляться пользователями, программами, автоматическими тестами и пакетными скриптами и разрабатываться и проверяться отдельно от будущих баз данных и устройств. Вместо «верха» и «низа» — «внутри» и «снаружи». Внутри — приложение и модель, снаружи — весь остальной мир, и общается он с приложением только через порты.
Адаптеры бывают двух видов. Ведущие (driving, primary) вызывают приложение: веб-обработчик, плановая задача, командная строка, тест. Ведомые (driven, secondary) вызываются приложением: хранилище, почта, платёжная система, часы. Шестиугольник — просто картинка, на которой хватает места для портов; число шесть ничего не значит.
| Порт Набора | Адаптер в бою | Адаптер в тестах |
|---|---|---|
CohortRepository | SqliteCohortRepository (или PostgreSQL) | InMemoryCohortRepository |
публикация событий: publish(events) | outbox и брокер сообщений (урок DDD.4.2) | RecordingPublisher — список в памяти |
| «есть ли у студента сертификат» для цены записи | клиент к Сертификации | словарь с ответами |
сам EnrollmentService (ведущий порт) | веб-обработчик, таймер 48 часов | тесты сценариев |
flowchart LR subgraph DRIVING["Ведущие адаптеры"] WEB["Веб-обработчик"] JOB["Таймер 48 часов"] TST["Тесты"] end subgraph CORE["Приложение"] AS["EnrollmentService"] M["Модель: Cohort, события,<br/>enrollment_price"] P1[/"порт CohortRepository"/] P2[/"порт publish(events)"/] end subgraph DRIVEN["Ведомые адаптеры"] SQL["SqliteCohortRepository"] MEM["InMemoryCohortRepository"] BUS["Шина событий, outbox"] end WEB --> AS JOB --> AS TST --> AS AS --> M AS --> P1 AS --> P2 SQL -.->|"реализует"| P1 MEM -.->|"реализует"| P1 BUS -.->|"реализует"| P2
Чистая и луковая архитектура
Заголовок раздела «Чистая и луковая архитектура»У той же идеи есть другие имена. Луковая архитектуралуковая архитектураEN onion architectureВариант Джеффри Палермо (2008): модель предметной области в центре, вокруг слои сервисов и инфраструктуры, зависимости направлены к центру. По сути та же идея, что у гексагональной и чистой архитектуры. Джеффри Палермо (2008) рисует модель в центре и слои вокруг неё. Чистая архитектурачистая архитектураEN Clean ArchitectureВариант Роберта Мартина (статья 2012 года, книга 2017): концентрические круги — сущности, сценарии использования, адаптеры интерфейсов, фреймворки и драйверы. Правило зависимостей: зависимости в исходном коде направлены только внутрь. Роберта Мартина (статья 2012 года, книга 2017-го) — концентрические круги: сущности, сценарии использования, адаптеры интерфейсов, фреймворки и драйверы — и правило зависимостей: зависимости в исходном коде направлены только внутрь, внутренний круг ничего не знает о внешних. Авторы «Cosmic Python» честно признают: споры о различиях есть, но все эти названия сводятся к инверсии зависимостей — модель не зависит от инфраструктуры.
| Классическая слоистая | Гексагональная, луковая, чистая | |
|---|---|---|
| Куда направлены зависимости | сверху вниз: модель зависит от слоя данных | внутрь: инфраструктура зависит от портов модели |
| Что в центре | база данных («фундамент») | модель предметной области |
| Тест правила | нужна база или моки ORM | фейк порта, база не нужна |
| Сменить базу | правка предметной области | новый адаптер + те же контрактные тесты |
Как тестировать модель без базы
Заголовок раздела «Как тестировать модель без базы»| Что тестируем | Чем | Сколько | Скорость |
|---|---|---|---|
| агрегаты, значения, доменные сервисы | обычные тесты pytest, без фейков | сотни | миллисекунды |
| прикладные сервисы | фейки портов: репозиторий в памяти, публикатор-список | десятки | миллисекунды |
| адаптеры | контрактные тесты на настоящей базе | один набор на порт | секунды |
| сквозные сценарии | несколько тестов через веб-интерфейс | единицы | секунды и дольше |
Фейк — рабочая подделка с настоящим, но упрощённым поведением. Мок (unittest.mock) проверяет другое — какие методы
и с какими аргументами вызвали, — и такие тесты часто ломаются при рефакторинге, даже когда поведение верно. Для своих портов берите фейки. Авторы
«Cosmic Python» добавляют правило «не мокайте то, чем не владеете»: вместо моков сессии SQLAlchemy — своя маленькая
абстракция, которую легко подделать честно.
Чтобы фейк не врал, его прогоняют теми же тестами, что и настоящий адаптер. Такие контрактные тесты описывают
поведение порта: что сохранили — то и достали, изменения без save теряются, параллельное сохранение отклоняется. В
pytest (урок 0.1.3) для этого есть фикстура — функция,
которая готовит объект для теста; с параметрами memory и sqlite каждый тест запускается дважды. Это та же страховка, что
тесты перед рефакторингом в уроке 11.1.2: вы меняете хранилище,
а контракт фиксирует поведение.
Структура Python-проекта по контекстам и слоям
Заголовок раздела «Структура Python-проекта по контекстам и слоям»В домашней работе модуля пакет enrollment/ плоский — шесть файлов, и для одного контекста это нормально. Когда
файлов становится больше, учебная школа раскладывается сначала по контекстам
(урок DDD.2.1), а внутри контекста — по слоям:
school/├── shared_kernel/ # общее ядро (урок DDD.2.2): Money, Period — маленькое и общее│ ├── money.py│ └── period.py├── enrollment/ # Набор (core): богатая модель, все три слоя│ ├── domain/ # модель: ничего не импортирует, кроме себя и shared_kernel│ │ ├── model.py # Cohort, Enrollment, исключения, фабрика│ │ ├── events.py│ │ ├── pricing.py # доменный сервис│ │ └── repository.py # порт CohortRepository (+ CohortNotFound, ConcurrencyError)│ ├── application/ # сценарии: импортирует domain│ │ └── services.py # EnrollmentService│ ├── infrastructure/ # адаптеры: импортируют domain и application│ │ ├── memory.py # InMemoryCohortRepository│ │ ├── sqlite.py # SqliteCohortRepository│ │ └── web.py # веб-обработчики — ведущий адаптер│ └── api.py # открытый интерфейс для других контекстов├── catalog/ # Каталог (supporting): простой CRUD, слои не нужны├── payment/ # Оплата (generic): тонкий адаптер и предохранительный слой (урок DDD.2.2)└── ...tests/├── enrollment/unit/ # модель и сценарии на фейках — быстрые├── enrollment/integration/ # контрактные тесты адаптеров└── e2e/ # несколько сквозных сценариевПравила дерева:
- сначала контексты, потом слои. Граница контекста видна в структуре папок, контекст легко вынести в отдельный сервис, а у разных контекстов разная глубина: Каталогу хватит CRUD (урок DDD.1.3);
- импорты только внутрь:
domainне импортируетapplicationиinfrastructure,applicationне импортируетinfrastructure; - другой контекст — только через его
api.pyили события, никогда через чужие классы и таблицы; - порт — в
domain, если он нужен модели и сценариям, как репозиторий. «Cosmic Python» кладёт порты рядом с адаптерами — это тоже рабочий вариант, если направление импортов соблюдается.
Правила, которые не проверяются, нарушаются, особенно в спешке. Архитектурный тест разбирает исходники модулем ast,
как линтер языка из урока DDD.1.2, и падает, если модель импортирует инфраструктуру. Для плоского пакета из практики:
"""Архитектурный тест: модель не зависит от инфраструктуры и прикладного слоя."""import astfrom pathlib import Path
PACKAGE = Path(__file__).parent.parent / "enrollment"DOMAIN = ["money.py", "events.py", "model.py", "pricing.py"]FORBIDDEN = {"sqlite3", "json", "sqlalchemy", "django", "requests", "repository", "sqlite_repository", "service"}
def imported_modules(path): """Все части имён модулей, которые импортирует файл: from .repository → {"repository"}.""" names = set() for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): if isinstance(node, ast.Import): names |= {alias.name for alias in node.names} elif isinstance(node, ast.ImportFrom) and node.module: names.add(node.module) return {part for name in names for part in name.split(".")}
def test_domain_does_not_import_infrastructure(): """Зависимости направлены к модели: money, events, model и pricing не знают о базе и сервисах.""" for name in DOMAIN: leaks = imported_modules(PACKAGE / name) & FORBIDDEN assert not leaks, f"{name} импортирует {sorted(leaks)} — модель должна обходиться без них"Для больших проектов есть готовые инструменты с «контрактами» слоёв, например пакет import-linter, но идея та же.
Чего избегать
Заголовок раздела «Чего избегать»Вот код, который встречается чаще, чем хотелось бы. В нём сразу три болезни:
class Cohort(Base): # модель — это строка таблицы ORM __tablename__ = "cohorts" cohort_id: Mapped[str] = mapped_column(primary_key=True) starts_at: Mapped[datetime] capacity: Mapped[int] price_kopecks: Mapped[int] # Money распалось на число без валюты enrollments: Mapped[list["EnrollmentRow"]] = relationship() # ленивая загрузка: SQL в любой момент
class EnrollmentManager: # все правила — в «сервисе-всезнайке» def enroll(self, cohort_id, student_id): cohort = self.session.get(Cohort, cohort_id) if datetime.now() >= cohort.starts_at: ... if len([e for e in cohort.enrollments if e.status != 2]) >= cohort.capacity: ... cohort.enrollments.append(EnrollmentRow(student_id=student_id, status=0)) self.session.commit() smtp.send(student_id, "Вы записаны") requests.post(PAYMENTS_URL, json={"order": cohort_id, "user": student_id})| Болезнь | Признаки в коде | Лечение |
|---|---|---|
| анемичная модельанемичная модель предметной областиEN anemic domain modelАнтишаблон, описанный Мартином Фаулером (2003): классы модели — мешки с данными без поведения, а все правила живут в сервисах. Выглядит как модель, но инварианты никто не защищает, правила дублируются. (Мартин Фаулер, 2003) | у Cohort только поля; все if о правилах — в сервисе; правило копируется в каждый сценарий; проверить его можно только через сервис с базой | правила — в методы агрегата (cohort.enroll), сервис только ведёт шаги |
| сервис-всезнайка | EnrollmentManager на сотни строк: правила, SQL, письма, HTTP, datetime.now(); меняется при любой задаче | правила — в модель, реакции — подписчики событий (урок DDD.3.3), SQL — адаптеры, время — параметром |
| утечка ORM в домен | модель наследует Base, ленивые связи делают запросы посреди метода, Money превратилось в колонку, session.commit() в сервисе | модель — обычные классы Python; отображение — в адаптере, как в SqliteCohortRepository, или императивное (раньше — «классическое») отображение SQLAlchemy, где ORM импортирует модель (глава 2 «Cosmic Python») |
Ещё три частые ошибки: репозиторий на каждую таблицу (запись меняют в обход потока), «универсальный» репозиторий с
тридцатью методами find_by_... для экранов (это модели чтения, урок DDD.4.1) и репозиторий, который отдаёт наружу
QuerySet или словари вместо агрегатов, — тогда ORM просачивается во все слои.
Анемичность не всегда грех. Для Каталога, где правил почти нет, модель-запись Django или SQLAlchemy — активная записьактивная запись (Active Record)EN Active RecordШаблон из той же книги Фаулера (2002), что и единица работы: объект — это строка таблицы, он сам себя загружает и сохраняет (модели Django ORM, Ruby on Rails). Удобен для CRUD и вспомогательных поддоменов, но связывает модель с базой. (Active Record) — экономит время. Болезнь начинается, когда так пишут основной поддомен со сложными правилами.
Практика
Заголовок раздела «Практика»Модель, порт, два адаптера и тесты
Заголовок раздела «Модель, порт, два адаптера и тесты»Возьмите проект из практики урока DDD.3.3 и добавьте в пакет enrollment/ файлы repository.py и
sqlite_repository.py из разделов выше, а в service.py — повтор при конфликте. Тесты положите в папку tests/:
school/├── enrollment/ # money, events, model, pricing, service — из урока DDD.3.3│ ├── repository.py # порт CohortRepository и InMemoryCohortRepository│ └── sqlite_repository.py└── tests/ ├── test_repositories.py ├── test_services.py └── test_architecture.pyКонтрактные тесты порта: фикстура repo с параметрами memory и sqlite запускает каждый тест на обоих адаптерах.
"""Контракт порта CohortRepository: одни и те же тесты для каждого адаптера."""import sqlite3from datetime import datetime
import pytest
from enrollment.model import Cohort, Coursefrom enrollment.money import Moneyfrom enrollment.repository import CohortNotFound, ConcurrencyError, InMemoryCohortRepositoryfrom enrollment.sqlite_repository import SqliteCohortRepository
NOW = datetime(2026, 10, 1, 10, 0)
@pytest.fixture(params=["memory", "sqlite"])def repo(request): if request.param == "memory": return InMemoryCohortRepository() return SqliteCohortRepository(sqlite3.connect(":memory:"))
def new_cohort(): course = Course("python", price=Money(3_000_000, "RUB")) return Cohort.open(course, datetime(2026, 11, 2, 10, 0), capacity=2, now=NOW)
def test_saved_cohort_comes_back_with_its_enrollments(repo): """Что сохранили, то и достали: записи, статусы, деньги, версия.""" repo.add(new_cohort()) cohort = repo.get("python-2026-11") cohort.enroll("anna", NOW) repo.save(cohort) again = repo.get("python-2026-11") assert [(e.student_id, e.status.value) for e in again.enrollments] == [("anna", "ожидает оплаты")] assert again.price == Money(3_000_000, "RUB") and again.version == 1
def test_changes_without_save_are_not_stored(repo): """Репозиторий отдаёт копию: изменения без save() в хранилище не попадают.""" repo.add(new_cohort()) repo.get("python-2026-11").enroll("anna", NOW) assert repo.get("python-2026-11").free_seats == 2
def test_concurrent_save_is_rejected(repo): """Два запроса загрузили одну версию — второе сохранение получает ConcurrencyError.""" repo.add(new_cohort()) first, second = repo.get("python-2026-11"), repo.get("python-2026-11") first.enroll("anna", NOW) second.enroll("boris", NOW) repo.save(first) with pytest.raises(ConcurrencyError): repo.save(second)
def test_unknown_cohort_is_not_found(repo): """Неизвестный номер — понятное исключение, а не None и не KeyError.""" with pytest.raises(CohortNotFound): repo.get("python-2099-01")Сценарии — на фейках. RacingRepository изображает гонку: перед первым сохранением «параллельный запрос» успевает
записать Бориса, и сервис должен повторить команду.
"""Сценарии Набора на фейках: без базы, без сети, за миллисекунды."""from datetime import datetime
import pytest
from enrollment.events import StudentEnrolledfrom enrollment.model import Cohort, Course, NoFreeSeatsfrom enrollment.money import Moneyfrom enrollment.repository import InMemoryCohortRepositoryfrom enrollment.service import EnrollmentService
NOW = datetime(2026, 10, 1, 10, 0)
class RecordingPublisher: """Фейк публикации: запоминает, что опубликовали.""" def __init__(self): self.published = []
def publish(self, events): self.published.extend(events)
def make_service(capacity=2, repo=None): repo = repo or InMemoryCohortRepository() course = Course("python", price=Money(3_000_000, "RUB")) repo.add(Cohort.open(course, datetime(2026, 11, 2, 10, 0), capacity, now=NOW)) publisher = RecordingPublisher() return EnrollmentService(repo, publisher), repo, publisher
def test_enroll_saves_cohort_and_publishes_event(): """Запись сохраняется, а СтудентЗаписан публикуется ровно один раз.""" service, repo, publisher = make_service() service.enroll("python-2026-11", "anna", NOW) assert repo.get("python-2026-11").free_seats == 1 assert [type(e) for e in publisher.published] == [StudentEnrolled]
def test_rejected_enrollment_saves_and_publishes_nothing(): """Отказ — не событие: при NoFreeSeats ничего не сохраняется и не публикуется.""" service, repo, publisher = make_service(capacity=1) service.enroll("python-2026-11", "anna", NOW) with pytest.raises(NoFreeSeats): service.enroll("python-2026-11", "boris", NOW) assert len(publisher.published) == 1 and repo.get("python-2026-11").version == 1
class RacingRepository(InMemoryCohortRepository): """Перед первым сохранением «параллельный запрос» успевает записать Бориса.""" def __init__(self): super().__init__() self.raced = False
def save(self, cohort): if not self.raced: self.raced = True rival = self.get(cohort.cohort_id) rival.enroll("boris", NOW) super().save(rival) super().save(cohort)
def test_service_retries_after_concurrency_conflict(): """Проиграв гонку, сервис перечитывает поток и повторяет команду: записаны оба.""" service, repo, publisher = make_service(repo=RacingRepository()) service.enroll("python-2026-11", "anna", NOW) assert sorted(e.student_id for e in repo.get("python-2026-11").enrollments) == ["anna", "boris"] assert [e.student_id for e in publisher.published] == ["anna"]Добавьте tests/test_architecture.py из раздела о структуре и запустите всё из папки school/:
$ python -m pytest -vtests/test_architecture.py::test_domain_does_not_import_infrastructure PASSEDtests/test_repositories.py::test_saved_cohort_comes_back_with_its_enrollments[memory] PASSEDtests/test_repositories.py::test_saved_cohort_comes_back_with_its_enrollments[sqlite] PASSEDtests/test_repositories.py::test_changes_without_save_are_not_stored[memory] PASSEDtests/test_repositories.py::test_changes_without_save_are_not_stored[sqlite] PASSEDtests/test_repositories.py::test_concurrent_save_is_rejected[memory] PASSEDtests/test_repositories.py::test_concurrent_save_is_rejected[sqlite] PASSEDtests/test_repositories.py::test_unknown_cohort_is_not_found[memory] PASSEDtests/test_repositories.py::test_unknown_cohort_is_not_found[sqlite] PASSEDtests/test_services.py::test_enroll_saves_cohort_and_publishes_event PASSEDtests/test_services.py::test_rejected_enrollment_saves_and_publishes_nothing PASSEDtests/test_services.py::test_service_retries_after_concurrency_conflict PASSED12 passed in 0.05s(Строки в квадратных скобках — параметры фикстуры; время на вашей машине будет другим, но счёт идёт на сотые доли
секунды. Если в папке school/ остался test_events.py из урока DDD.3.3, pytest соберёт и его — тестов будет 15.) Ни один тест сценариев не открыл базу, а SQLite проверен тем же контрактом, что и фейк.
Упражнения
Заголовок раздела «Упражнения»1. Сломайте фейк (10 минут). Уберите copy.deepcopy из InMemoryCohortRepository.get — пусть отдаёт сам
сохранённый объект. Какие тесты упадут и почему это ошибки, которые фейк скрыл бы от тестов сценариев?
Разбор
Падают три теста, все — на адаптере memory: test_changes_without_save_are_not_stored (изменения без save сразу
попали в «хранилище»), test_concurrent_save_is_rejected (оба запроса получили один объект, гонку не видно) и
test_service_retries_after_concurrency_conflict. С таким фейком сервис, забывший вызвать save, прошёл бы все
тесты и сломался бы только в бою. Ровно поэтому фейк проверяют контрактом вместе с настоящим адаптером.
2. Новый метод порта (15 минут). Плановой задаче нужны все потоки, где есть неоплаченные записи старше 48 часов, а
странице курса — «свободные места по всем потокам». Добавлять ли оба метода в CohortRepository?
Разбор
Первый — да, это запрос модели для команды: задача найдёт потоки и вызовет у каждого отмену по таймеру. Назовите его
словами домена, например with_unpaid_enrollments_older_than(moment), и добавьте в контрактные тесты. Второй — нет:
экрану не нужны агрегаты целиком, ему нужна таблица чисел. Это модель чтения, которую строят отдельно, часто из
событий (урок DDD.4.1). Если тянуть такие запросы в репозиторий, он
быстро превратится в «универсальный» с тридцатью методами.
3. Адаптер с другим хранением (30 минут, по желанию). Напишите JsonFileCohortRepository, который хранит каждый
поток в отдельном файле <номер>.json, и добавьте его третьим параметром фикстуры. Пройдут ли контрактные тесты без
изменений? Где придётся подумать о версии?
Подсказка
Отображение в JSON у вас уже есть — функции _to_json и _from_json из sqlite_repository.py; вынесите их в общий
модуль адаптеров. Версию храните в том же файле и сравнивайте перед записью. Между чтением и записью файла может
вклиниться другой процесс — в учебном адаптере это допустимо, а в бою проверку и запись делают атомарно, как
UPDATE ... WHERE version = ? в SQL.
Проверь себя
Заголовок раздела «Проверь себя»- Репозиторий — иллюзия коллекции агрегатов:
add,get,saveсловами модели, один репозиторий на тип агрегата. Порт объявлен рядом с моделью, адаптеры (память, SQLite) — в инфраструктуре; отображение в строки живёт в адаптере, восстановление идёт через конструктор агрегата. - Оптимистическая блокировка в базе —
UPDATE ... WHERE version = ?, в фейке — сравнение версий; конфликт прикладной сервис решает повтором команды на свежих данных. Несколько изменений в одной транзакции фиксирует единица работы. - Слоистая архитектура делает модель зависимой от базы. Инверсия зависимостей разворачивает стрелку: в гексагональной, луковой и чистой архитектуре модель в центре, а адаптеры зависят от её портов.
- Модель и сценарии тестируют без базы — обычными тестами и фейками портов; фейк держат честным контрактными тестами вместе с настоящим адаптером; архитектурный тест следит за направлением импортов.
- Проект раскладывают сначала по контекстам, потом по слоям. Чего избегать: анемичной модели, сервисов-всезнаек, утечек ORM в домен, репозиториев на таблицу и «универсальных» репозиториев для экранов.
Дальше: урок DDD.4.1 — CQRS и event sourcing. Разделим модели записи и чтения, построим «свободные места по потокам» из событий и восстановим поток свёрткой его журнала.