Репозитории, единица работы и персистентность без протечек

В прошлой статье мы собрали агрегат — кусочек модели, который сам защищает себя от невалидных состояний. Он живёт в памяти, проверяет инварианты и не знает ни слова про базу данных. Но программа перезапускается, а заказ должен остаться.

Вопрос этой статьи звучит обманчиво просто:

как положить агрегат в хранилище и достать обратно, не протащив хранилище внутрь домена?

Наивный ответ («просто добавь ORM») ломается ровно там, где начинается настоящая система: при конкурентных изменениях, больших графах объектов, смене схемы БД, тестировании и попытке отделить чтение от записи.

1. Интуиция: иллюзия бесконечной памяти

Представьте, что оперативная память бесконечна и никогда не гаснет. Тогда персистентность не нужна: заказы просто лежат в коллекции.

orders = set()                                     # все заказы мира

order = Order.create(customer_id, items)
orders.add(order)                                  # сохранили
o = next(x for x in orders if x.id == order_id)    # достали
o.cancel(reason="передумал")                       # изменили — изменение уже "сохранено"

Это идеальный API работы с доменом: коллекция объектов. Ни транзакций, ни SQL, ни мэппинга — разработчик думает только о бизнес-правилах. Реальность другая: память конечна, процесс падает, объектов миллионы, доступ идёт из десяти реплик. Но интерфейс можно оставить прежним. Это и есть репозиторий.

Repository — объект, имитирующий коллекцию агрегатов в памяти и скрывающий за собой реальное хранилище (Evans, гл. 6; Fowler, PoEAA, Repository).

Проверочный критерий качества: прикладной код читается так, будто базы данных не существует. Если в сценарии встречаются session.flush(), .query().join(), SELECT FOR UPDATE или Include(x => x.Lines) — абстракция протекла.

2. Репозиторий — на агрегат, а не на таблицу

Частая ошибка: OrderRepository, OrderLineRepository, AddressRepository — по репозиторию на класс. Это признак мышления таблицами. Правило вытекает прямо из определения агрегата:

Один репозиторий — на один корень агрегата. Внутренние сущности агрегата своего репозитория не имеют.

Почему: агрегат — единица транзакционной согласованности. Если позицию заказа можно загрузить и сохранить отдельно от заказа, инвариант «сумма позиций ≤ кредитного лимита» больше никем не охраняется. Репозиторий на дочернюю сущность — дыра в границе агрегата. Отсюда следствия: репозиторий всегда возвращает целый агрегат, сохраняет его целиком и атомарно, а число репозиториев равно числу типов агрегатов (обычно 5–30 на контекст, не 200).

На схеме OrderLine живёт в отдельной таблице, но своего репозитория не имеет; Value Object Address растворился в колонках корневой таблицы; CustomerId — просто идентификатор, внешний агрегат грузится отдельно. Схема БД и объектная модель связаны, но не тождественны.

Интерфейс репозитория — часть доменного слоя, реализация — часть инфраструктуры. Это Dependency Inversion: домен объявляет, что ему нужно, инфраструктура подчиняется.

Стрелка реализации направлена из инфраструктуры в домен — это и есть «зависимости внутрь» из обзора трека.

3. Unit of Work: транзакция как объект

Репозитория недостаточно. Сценарий «перенести деньги» трогает два агрегата, и оба изменения должны примениться атомарно. Если транзакцией управляет репозиторий (repo.save() = COMMIT), атомарность двух сохранений недостижима. Значит, транзакция — отдельная сущность.

Unit of Work отслеживает все объекты, затронутые бизнес-операцией, и координирует запись изменений и разрешение конфликтов (Fowler, PoEAA).

Практически UoW делает три вещи:

1. Граница транзакции — begin на входе в сценарий, commit/rollback на выходе.
2. Identity Map — гарантия, что один агрегат в рамках операции представлен одним объектом в памяти (иначе два экземпляра Order#42 разойдутся в состоянии).
3. Реестр изменений — что добавлено, изменено, удалено; на commit превращается в минимальный набор INSERT/UPDATE/DELETE.

Обратите внимание на переход Clean → Persisted без запросов: хороший UoW не пишет то, что не менялось.

import abc
from sqlalchemy.orm import Session, sessionmaker


class AbstractUnitOfWork(abc.ABC):
    """Интерфейс живёт в прикладном слое: никаких импортов SQLAlchemy."""
    orders: "OrderRepository"

    def __enter__(self): return self

    def __exit__(self, *args):
        self.rollback()          # не вызвали commit — значит откат, безопасное поведение

    @abc.abstractmethod
    def commit(self) -> None: ...

    @abc.abstractmethod
    def rollback(self) -> None: ...

    def collect_new_events(self):
        """События со всех затронутых агрегатов — понадобится в статье 05."""
        for agg in self.orders.seen:
            while agg.events:
                yield agg.events.pop(0)


class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
    def __init__(self, session_factory: sessionmaker):
        self._session_factory = session_factory

    def __enter__(self):
        self.session: Session = self._session_factory()
        self.orders = SqlAlchemyOrderRepository(self.session)
        return self

    def __exit__(self, *args):
        self.rollback()
        self.session.close()

    def commit(self):   self.session.commit()
    def rollback(self): self.session.rollback()

Прикладной сценарий тогда читается как бизнес-текст:

def add_item_to_order(cmd: AddItem, uow: AbstractUnitOfWork) -> None:
    with uow:
        order = uow.orders.get(cmd.order_id)
        if order is None:
            raise OrderNotFound(cmd.order_id)
        order.add_line(sku=cmd.sku, qty=cmd.qty, price=cmd.price)   # ВСЯ логика внутри агрегата
        uow.commit()

Ни session, ни flush, ни SELECT — три строки, которые можно показать аналитику. Заметьте: add() после get() не вызывается, UoW сам заметит изменение.

4. Полный сценарий: кто с кем разговаривает

Четыре вывода из диаграммы: транзакция начинается и заканчивается в прикладном слое, не в контроллере и не в репозитории; домен вообще не разговаривает с БД, он только принимает решения; проверка версии — часть UPDATE, а не отдельный SELECT (иначе гонка возвращается); ошибка домена (422) и ошибка конкурентности (409) — разные вещи, первую повторять бессмысленно, вторую нужно.

5. Мэппинг: три способа и их цена

5.1. Active Record — модель наследует ORM

class Order(Base):          # Django Model, Eloquent, ActiveRecord
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    def cancel(self): ...

Быстро и дёшево, отлично для CRUD. Но домен наследует инфраструктуру: появляются save(), delete(), ленивые связи, требование живой сессии. Проверить бизнес-логику без базы невозможно. Для сложного домена — путь к протечкам; для админки и простого CRUD — правильный выбор (о том, когда DDD не нужен, — в обзоре).

5.2. Data Mapper с «классическим» мэппингом — домен чист

SQLAlchemy умеет мэппить обычные классы, ничего не зная о них заранее (imperative mapping) — редкая, но самая ценная для DDD возможность.

# domain/order.py — ноль импортов из инфраструктуры
from dataclasses import dataclass
from decimal import Decimal


@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str


@dataclass
class OrderLine:
    sku: str
    qty: int
    price: Money


class Order:
    MAX_LINES = 50

    def __init__(self, id: str, customer_id: str, lines: list[OrderLine] | None = None):
        self.id, self.customer_id = id, customer_id
        self.lines = lines or []
        self.status = "draft"
        self.version = 0
        self.events: list = []

    def add_line(self, sku: str, qty: int, price: Money) -> None:
        if self.status != "draft":
            raise DomainError("нельзя менять состав подтверждённого заказа")
        if len(self.lines) >= self.MAX_LINES:
            raise DomainError(f"в заказе не может быть больше {self.MAX_LINES} позиций")
        self.lines.append(OrderLine(sku, qty, price))
        self.events.append(LineAdded(self.id, sku, qty))

# infrastructure/orm.py — вся грязь здесь
from sqlalchemy import Table, Column, String, Integer, Numeric, ForeignKey, MetaData
from sqlalchemy.orm import registry, relationship, composite

metadata, mapper_registry = MetaData(), registry()

orders = Table("orders", metadata,
    Column("id", String(36), primary_key=True),
    Column("customer_id", String(36), nullable=False, index=True),
    Column("status", String(16), nullable=False),
    Column("version", Integer, nullable=False, server_default="0"))

order_lines = Table("order_lines", metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("order_id", String(36), ForeignKey("orders.id", ondelete="CASCADE"), nullable=False),
    Column("sku", String(64), nullable=False), Column("qty", Integer, nullable=False),
    Column("price_amount", Numeric(18, 2)), Column("price_currency", String(3)))


def start_mappers() -> None:
    """Вызывается один раз при старте приложения."""
    line_mapper = mapper_registry.map_imperatively(OrderLine, order_lines, properties={
        # Value Object собирается из двух колонок и обратно
        "price": composite(Money, order_lines.c.price_amount, order_lines.c.price_currency),
    })
    mapper_registry.map_imperatively(Order, orders,
        properties={"lines": relationship(line_mapper,
            cascade="all, delete-orphan",   # позиции живут только внутри агрегата
            lazy="selectin")},              # грузим одним доп. запросом, без N+1
        version_id_col=orders.c.version)    # оптимистичная блокировка «из коробки»

Что мы получили: домен тестируется без базы за микросекунды; схему БД можно менять, правя только orm.py; cascade="all, delete-orphan" кодирует доменное правило «позиция не существует вне заказа»; version_id_col даёт оптимистичную блокировку без единой строки в домене. Это подход из Architecture Patterns with Python Персиваля и Грегори — лучшего бесплатного источника по теме.

5.3. Ручной мэппер — максимальная свобода

Когда ORM мешает (event sourcing, экзотическое хранилище, агрегат из трёх источников), мэппинг пишут руками:

class SqlOrderRepository:
    def __init__(self, conn):
        self._conn = conn
        self.seen: set[Order] = set()      # для сбора событий и UoW

    def get(self, order_id: str) -> Order | None:
        head = self._conn.execute(
            "SELECT id, customer_id, status, version FROM orders WHERE id = %s", (order_id,)).fetchone()
        if head is None:
            return None
        rows = self._conn.execute(
            "SELECT sku, qty, price_amount, price_currency FROM order_lines WHERE order_id = %s",
            (order_id,)).fetchall()
        order = Order(head.id, head.customer_id,
                      [OrderLine(r.sku, r.qty, Money(r.price_amount, r.price_currency)) for r in rows])
        order.status, order.version = head.status, head.version   # version — техническое поле
        self.seen.add(order)
        return order

    def save(self, order: Order) -> None:
        cur = self._conn.execute(
            "UPDATE orders SET status = %s, version = version + 1 WHERE id = %s AND version = %s",
            (order.status, order.id, order.version))
        if cur.rowcount == 0:
            raise ConcurrencyError(order.id)                      # кто-то опередил
        # позиции: простейшая стратегия — снести и записать заново (агрегат мал)
        self._conn.execute("DELETE FROM order_lines WHERE order_id = %s", (order.id,))
        self._conn.executemany(
            "INSERT INTO order_lines (order_id, sku, qty, price_amount, price_currency) "
            "VALUES (%s, %s, %s, %s, %s)",
            [(order.id, l.sku, l.qty, l.price.amount, l.price.currency) for l in order.lines])
        order.version += 1
        self.seen.add(order)

Сложность. get — O(1 + k) запросов (обычно 2) и O(k) памяти, где k — число позиций. save в стратегии delete-and-reinsert — O(k) записей независимо от того, изменилась одна позиция или все. Осознанный размен: простота кода против лишних записей. Он допустим, пока k мало (десятки) — ещё один аргумент за маленькие агрегаты. При k в тысячах нужна дельта-запись, и тут ORM с dirty tracking выигрывает.

6. Схема данных агрегата

Почему нет FK orders.customer_id → customers.id. Ссылка между агрегатами — это ссылка по идентификатору, а не по объекту. Жёсткий FK связывает жизненные циклы и мешает разнести агрегаты по разным базам, когда придёт время. Целостность обеспечивается бизнес-правилом (при создании заказа проверяем существование покупателя). Решение спорное: в монолите FK дёшев и ловит баги. Разумный компромисс — оставлять FK внутри одного ограниченного контекста и никогда не ставить между контекстами.

Почему ON DELETE CASCADE внутри агрегата. Позиция не существует без заказа — тот случай, когда констрейнт БД точно отражает доменное правило.

Почему outbox тут же. События пишутся в той же транзакции, что и изменение агрегата, иначе возможно «заказ сохранён, событие потеряно». Механику разберём в следующей статье, но место резервируем сразу — это влияет на дизайн UoW.

7. Конкурентность: главная причина, почему всё сложно

Два пользователя одновременно правят заказ. Оба прочитали version = 7, оба пишут. Без защиты второй молча затирает первого — классическое потерянное обновление.

Оптимистичная блокировка

UPDATE orders
   SET status = 'confirmed',
       version = version + 1
 WHERE id = $1
   AND version = $2;      -- версия, прочитанная в начале транзакции
-- rowcount = 0  →  кто-то опередил  →  ConcurrencyError

Деталь, которую путают: версия инкрементируется при любом изменении внутри агрегата, включая изменения дочерних сущностей. Добавили позицию — версия корня выросла. Так граница агрегата превращается в границу блокировки: конфликтуют только те операции, которые действительно конфликтуют по бизнесу.

• Плюсы: нет блокировок в БД, отлично масштабируется, работает через границы транзакций (можно проверить версию, пришедшую с фронтенда через час — «offline lock»).
• Минусы: конфликт обнаруживается поздно, работа теряется, нужен повтор.
• Когда: конфликты редки (>95 % операций не конфликтуют) — подавляющее большинство систем.

Пессимистичная блокировка

SELECT * FROM orders WHERE id = $1 FOR UPDATE;          -- другие ждут
SELECT * FROM orders WHERE id = $1 FOR UPDATE NOWAIT;   -- мгновенная ошибка вместо очереди

Плюс — конфликт исключён, работа не теряется. Минусы — очереди, дедлоки, деградация под нагрузкой, невозможность держать блокировку между HTTP-запросами. Уместна для коротких операций с высокой конкуренцией на одну строку: списание со счёта, выдача последнего товара со склада.

Повтор как часть контракта

Ретрай — это пересчёт всего сценария на свежих данных, а не повторная запись старого объекта:

def with_retry(fn, attempts: int = 3, base_delay: float = 0.02):
    """Экспоненциальная задержка + джиттер против 'стада' повторов."""
    for attempt in range(attempts):
        try:
            return fn()                    # ВЕСЬ сценарий: get → изменить → commit
        except ConcurrencyError:
            if attempt == attempts - 1:
                raise
            time.sleep(base_delay * (2 ** attempt) + random.uniform(0, base_delay))

Обязательное условие корректности — идемпотентность на уровне бизнеса: если сценарий списывает деньги, повтор должен опираться на ключ идемпотентности команды, иначе спишем трижды. В PostgreSQL при SERIALIZABLE аналогично нужно ловить 40001 serialization_failure и повторять — это прямо описано в документации.

Практический вывод: не полагайтесь на уровень изоляции как на защиту инвариантов. Инвариант охраняет агрегат + версия, изоляция — страховка второго эшелона. Систему, корректность которой держится только на SERIALIZABLE, невозможно перенести в распределённую среду.

8. Чтение: почему репозиторий — плохой инструмент для отчётов

Здесь ломается больше всего проектов. Агрегат оптимизирован под изменение: маленький, целостный, загружается целиком. Экран «список заказов с именем клиента, суммой и статусом доставки» требует другого: плоской выборки из пяти таблиц без единого инварианта.

Обслуживание такого экрана через репозиторий даёт: методы get_orders_with_customer_and_shipping_and_totals(...) (репозиторий превращается в DAO); загрузку 500 агрегатов целиком ради 20 строк таблицы (O(n·k) объектов вместо O(n) строк); N+1 запросов и «почему страница грузится 4 секунды».

Решение — разделить чтение и запись (первый шаг к CQRS, см. трек архитектурных паттернов):

def confirm_order(cmd, uow):                 # Запись — через репозиторий и агрегат
    with uow:
        order = uow.orders.get(cmd.order_id)
        order.confirm()                      # инварианты
        uow.commit()


def list_orders_view(conn, customer_id: str, limit: int = 20) -> list[dict]:
    """Чтение — плоский SQL в read-модель, никаких агрегатов."""
    return conn.execute("""
        SELECT o.id, o.status, o.total_amount, c.name AS customer_name, COUNT(l.id) AS line_count
          FROM orders o
          JOIN customers c ON c.id = o.customer_id
          LEFT JOIN order_lines l ON l.order_id = o.id
         WHERE o.customer_id = %s
         GROUP BY o.id, c.name
         ORDER BY o.created_at DESC
         LIMIT %s""", (customer_id, limit)).fetchall()

Да, это «нарушение чистоты»: SQL прямо в прикладном слое. Но чтение не имеет инвариантов, защищать здесь нечего.

Команды идут через агрегаты и репозитории. Запросы — через read-модели или прямой SQL. Смешивать их в одном интерфейсе — источник и медленных страниц, и раздутых репозиториев.

Иногда критерий выборки действительно доменный («просроченный заказ»). Тогда его формулируют как Specification (Evans & Fowler) с двумя представлениями — is_satisfied_by() для тестов и in-memory и to_sql() для базы, — а репозиторий её транслирует. Плюс: критерий один и живёт в домене. Минус: двойная реализация может разойтись, поэтому покрывайте обе одним набором тестов. Не вводите спецификации, пока критерий не задублировался в трёх местах.

9. Тестирование: главный дивиденд абстракции

Репозиторий окупается ровно тогда, когда нужны 300 быстрых тестов бизнес-логики.

class FakeOrderRepository:
    """In-memory дублёр: 12 строк вместо поднятого PostgreSQL."""

    def __init__(self, orders: list[Order] | None = None):
        self._orders = {o.id: o for o in (orders or [])}
        self.seen: set[Order] = set()

    def get(self, order_id):
        o = self._orders.get(order_id)
        if o:
            self.seen.add(o)
        return o

    def add(self, order):
        self._orders[order.id] = order
        self.seen.add(order)


class FakeUnitOfWork(AbstractUnitOfWork):
    def __init__(self):
        self.orders, self.committed = FakeOrderRepository(), False

    def commit(self):   self.committed = True
    def rollback(self): pass


def test_нельзя_менять_подтверждённый_заказ():
    uow = FakeUnitOfWork()
    order = Order("o-1", "c-1")
    order.add_line("SKU-1", 1, Money(Decimal("100"), "RUB"))
    order.confirm()
    uow.orders.add(order)

    with pytest.raises(DomainError):
        add_item_to_order(AddItem("o-1", "SKU-2", 1, Money(Decimal("50"), "RUB")), uow)
    assert uow.committed is False        # неудачный сценарий не коммитит

Тест выполняется за микросекунды и не требует докера. Но фейк лжёт: он не проверит каскадное удаление, констрейнты, конфликт версий и особенности мэппинга. Отсюда обязательная вторая линия — контрактные тесты репозитория: один и тот же набор тестов прогоняется против фейка и против настоящей базы.

@pytest.fixture(params=["fake", "postgres"])
def repo(request, pg_session):
    return FakeOrderRepository() if request.param == "fake" else SqlAlchemyOrderRepository(pg_session)


def test_сохранённый_заказ_читается_с_позициями(repo, uow_commit):
    order = Order("o-1", "c-1")
    order.add_line("SKU-1", 2, Money(Decimal("100"), "RUB"))
    repo.add(order); uow_commit()

    loaded = repo.get("o-1")
    assert len(loaded.lines) == 1 and loaded.lines[0].qty == 2

Настоящую базу поднимают через Testcontainers — дороже (секунды вместо микросекунд), поэтому таких тестов десятки, а не сотни.

SQLite вместо PostgreSQL в контрактных тестах — распространённая, но опасная экономия: расходятся типы, поведение при конфликтах, RETURNING, JSON-операторы и уровни изоляции. Тестируйте на той СУБД, что стоит в проде.

10. Типичные ошибки

1. repo.save() в середине сценария. Три вызова = три коммита, атомарность потеряна. Коммит — прерогатива UoW, один на сценарий.
2. Репозиторий, возвращающий DTO или ORM-модель. Логика неизбежно уезжает в прикладной слой: анемичная модель плюс лишний слой абстракции — худшее из двух миров.
3. IRepository<T> с 20 generic-методами. GetAll(), Find(Expression<Func<T,bool>>), Query() — это обёртка над ORM с нулевой семантикой: LINQ-выражение утекает в домен, а границы не выражены. Репозиторий должен иметь мало методов с доменными именами: get, add, find_overdue.
4. Ленивая загрузка наружу. order.customer.name внутри доменного метода — скрытый SELECT; в цикле по 100 заказам это 100 запросов. Внутри агрегата грузите жадно (selectin/Include), между агрегатами не держите ссылок на объекты вообще.
5. Агрегат, загружающий полтаблицы. Customer со всеми заказами за пять лет читается целиком ради одного нового заказа. Это не проблема репозитория, а неверная граница агрегата (см. тактические блоки). Если get() стабильно тянет мегабайты — режьте агрегат.
6. Транзакция на два агрегата. Технически возможно, но это скрытая связность: вы навсегда привязали два агрегата к одной базе. Правило «одна транзакция — один агрегат» отдаёт согласованность второго на откуп доменным событиям.
7. Бизнес-логика в триггерах и хранимках. Уезжает туда, где её не видно, не покрыть юнит-тестами и не увидеть в code review. Констрейнты (NOT NULL, CHECK, UNIQUE) — да, это защита данных. Бизнес-решения — нет.
8. Отсутствие версии. «У нас конфликты редки» — до первого двойного списания. Колонка version стоит 4 байта и одну строку в мэппинге; добавляйте сразу.
9. Кэш поверх репозитория без инвалидации. Почти всегда даёт устаревшие данные при записи. Кэшируйте read-модели, не агрегаты.

11. Как это выглядит в проде

.NET / EF Core. DbContext уже является и Unit of Work, и набором репозиториев — официальная позиция Microsoft в руководстве по DDD-микросервисам. Собственный репозиторий поверх пишут не ради абстракции от БД, а ради выражения границ агрегата — чтобы никто не сделал context.OrderLines.Where(...) в обход корня.

public class OrderRepository : IOrderRepository
{
    private readonly OrderingContext _context;
    public IUnitOfWork UnitOfWork => _context;              // коммитит вызывающий сценарий

    public OrderRepository(OrderingContext context) => _context = context;

    public Task<Order?> GetAsync(Guid id, CancellationToken ct) =>
        _context.Orders
                .Include(o => o.OrderItems)                 // агрегат грузится целиком
                .FirstOrDefaultAsync(o => o.Id == id, ct);

    public Order Add(Order order) => _context.Orders.Add(order).Entity;
}

Приватные поля-коллекции с публичным IReadOnlyCollection (backing fields) позволяют закрыть коллекцию позиций от внешних изменений — редкий случай, когда ORM прямо поддерживает инкапсуляцию агрегата.

JVM / Spring. JpaRepository<T, ID> с 40 унаследованными методами — это DAO, а не репозиторий в смысле DDD. Практика зрелых команд: узкий интерфейс в домене, JPA-реализация в инфраструктуре, обязательный @Version и осторожность с @OneToMany(fetch = LAZY), чтобы не ловить LazyInitializationException вне транзакции. Отдельный вариант — Spring Data JDBC, спроектированный прямо вокруг агрегата: никакого lazy loading, сохранение агрегата целиком.

Go. Идиома — интерфейс в пакете домена (type Repository interface { Get(ctx, id) (*Order, error); Save(ctx, *Order) error }), реализация рядом с БД, транзакция передаётся через контекст или обёртку WithTx(ctx, func(tx) error). Подробнее о структуре проектов — в треке по Go.

Документные и event-sourced хранилища. Если агрегат маленький и всегда читается целиком, документная БД или колонка jsonb — идеальное совпадение с моделью: один агрегат = один документ, атомарность бесплатна, мэппинг сводится к сериализации. Ограничение — запросы по внутренностям агрегата и версионирование формата документа. При event sourcing get() читает поток событий и сворачивает их (O(m) от длины потока, лечится снапшотами каждые N событий), save() дописывает события с проверкой ожидаемой версии потока. Интерфейс для прикладного слоя при этом остаётся тем же — лучшее доказательство ценности абстракции.

12. Сколько это стоит: честные trade-offs

Честно про «замену БД»: за карьеру это происходит один-два раза. Настоящая ценность репозитория — не переносимость, а тестируемость и явность границ агрегата. Если вы вводите репозиторий ради гипотетического переезда с PostgreSQL на MongoDB — вы вводите его по неверной причине и, скорее всего, переусложните.

Репозиторий и UoW не нужны: в CRUD-сервисе без инвариантов (берите Active Record); в скриптах, ETL и аналитике (там царит SQL, и это правильно); в прототипе на три недели; и пока команда не договорилась о модели домена — абстракция над несуществующей моделью есть чистые накладные расходы.

13. Мини-итог

• Репозиторий — иллюзия коллекции агрегатов в памяти. Критерий качества: прикладной код читается так, будто базы нет.
• Репозиторий — на корень агрегата, а не на таблицу и не на класс; возвращает и сохраняет агрегат целиком.
• Интерфейс в домене, реализация в инфраструктуре: зависимости направлены внутрь.
• Unit of Work владеет транзакцией, Identity Map и реестром изменений. Один коммит на сценарий.
• Мэппинг — отдельное решение: Active Record для CRUD, Data Mapper для доменных сервисов, ручной мэппер для экзотики.
• Конкурентность решается колонкой version плюс ретраем всего сценария; пессимистичная блокировка — для коротких горячих операций.
• Чтение — мимо репозитория: отчёты и списки берут плоский SQL или read-модель.
• Тесты: быстрые фейки для логики плюс контрактные тесты против настоящей СУБД для мэппинга.
• Признак протечки: flush, Include, join или LazyInitializationException в прикладном или доменном коде.

Источники

• Eric Evans. Domain-Driven Design (2003), гл. 6 «The Life Cycle of a Domain Object»; выжимка — DDD Reference.
• Martin Fowler. Patterns of Enterprise Application Architecture — Repository, Unit of Work, Data Mapper, Identity Map, Optimistic Offline Lock.
• Evans & Fowler. Specifications (PDF).
• Percival & Gregory. Architecture Patterns with Python, гл. 2 «Repository» и гл. 6 «Unit of Work» — бесплатно на cosmicpython.com.
• Vaughn Vernon. Implementing Domain-Driven Design, гл. 12 «Repositories».
• Microsoft. Infrastructure persistence layer design и реализация на EF Core.
• SQLAlchemy. Imperative (classical) mapping, Configuring a Version Counter.
• PostgreSQL. Transaction Isolation — почему при SERIALIZABLE ретраи обязательны.
• Spring. Spring Data JDBC — Aggregate Roots.

Что дальше

Мы научились атомарно сохранять один агрегат. Но бизнес-процесс редко им заканчивается: заказ подтверждён — надо зарезервировать склад, начислить бонусы, уведомить клиента. Тянуть это в одну транзакцию нельзя, а терять — тем более. Разберём доменные события, шаблон Outbox, гарантии доставки и способы интеграции контекстов без превращения системы в распределённый монолит.

Читайте: Доменные события и интеграция контекстов.