Паттерн Unit of Work для Python-сервисов: транзакции, репозитории и единая точка коммита

В реальных Python-сервисах почти всегда рано или поздно возникает одна и та же проблема: изменения, которые логически относятся к одной бизнес-операции, затрагивают несколько таблиц и/или несколько репозиториев. И если каждый репозиторий коммитит сам «как умеет», то вы получаете труднообнаружимые ошибки: частично записанные данные, «висящие» статусы, рассинхрон между агрегатами, а главное — отсутствие единой точки, где можно гарантировать атомарность.

Паттерн Unit of Work (UoW) решает это архитектурно: все изменения на время бизнес-сценария выполняются в рамках одной транзакции, а коммит/роллбек контролируются в одном месте. Репозитории перестают управлять транзакциями и становятся тонкими слоями доступа к данным.

Ниже разберём практическую реализацию UoW для Python-сервисов с фокусом на:

• границы транзакции и «единая точка коммита»;
• обработку ошибок и корректный роллбек;
• организацию репозиториев так, чтобы они не знали про транзакции;
• тестирование без хрупких моков, через интеграционные подходы.

Почему транзакции “расползаются” по коду

Частый анти-паттерн выглядит так:

• сервис вызывает repo_a.update(...) — и внутри repo_a делает commit;
• потом сервис вызывает repo_b.create(...) — и внутри repo_b делает commit.

Если во втором репозитории что-то падает, первый коммит уже случился. Даже если вы затем сделаете «компенсацию» (например, откатите изменения вручную), вы столкнётесь с проблемами:

• бизнес-логика становится сложнее и менее надёжной;
• появляются гонки и нестабильные состояния;
• диагностика инцидентов превращается в расследование цепочки частичных изменений.

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

Unit of Work — способ организовать это в коде.

Unit of Work: идея и контракт

Unit of Work — это объект, который:

1. создаёт или получает контекст БД (часто — сессия/connection);
2. открывает транзакцию;
3. даёт репозиториям доступ к этой сессии;
4. в конце принимает решение:
   • если всё прошло успешно — делает commit;
   • если произошла ошибка — делает rollback.

Обычно контракт реализуют через контекстный менеджер:

with uow:
    # действия репозиториев
# здесь либо commit, либо rollback

Важно: UoW отвечает за коммит, а репозитории — только за операции чтения/записи через переданный контекст.

Архитектурный скелет: транзакция, репозитории и единая точка коммита

Ниже — рабочий каркас на основе SQLAlchemy 2.x (sync). Он демонстрирует ключевые элементы: границы транзакции, обработку ошибок и отсутствие «commit внутри репозитория».

Слой данных: модели (минимально)

Предположим, есть две сущности: Account и LedgerEntry. Бизнес-операция: перевод средств.

from dataclasses import dataclass

# Для примера импортируем ORM-модели условно.
# В реальном проекте это будут SQLAlchemy модели.

Репозитории: без управления транзакциями

Репозиторий получает session (или uow.session) и использует его. Никаких коммитов.

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

class AccountRepository:
    def __init__(self, session):
        self.session = session

    def get_by_id(self, account_id: int):
        stmt = select(Account).where(Account.id == account_id)
        return self.session.execute(stmt).scalar_one()

    def decrement(self, account_id: int, amount: int):
        acc = self.get_by_id(account_id)
        acc.balance -= amount
        # Никаких commit — только изменение состояния ORM-объекта.
        return acc

    def increment(self, account_id: int, amount: int):
        acc = self.get_by_id(account_id)
        acc.balance += amount
        return acc


class LedgerRepository:
    def __init__(self, session):
        self.session = session

    def add_entry(self, account_id: int, amount: int, kind: str, reference_id: str):
        entry = LedgerEntry(
            account_id=account_id,
            amount=amount,
            kind=kind,
            reference_id=reference_id,
        )
        self.session.add(entry)
        return entry

Подводный камень: репозиторий может делать flush() — это полезно для получения ID и проверки ограничений — но это не коммит. Коммит — ответственность UoW. Если вы начнёте «коммитить» во внутренностях — атомарность пропадёт.

Реализация Unit of Work

Ниже — UoW как контекстный менеджер с управлением коммитом/роллбеком. Поддержим:

• коммит только при отсутствии исключений;
• роллбек при любой ошибке;
• корректное закрытие сессии;
• аккуратную стратегию обработки «известных» ошибок (например, IntegrityError).

Базовый UoW

from contextlib import AbstractContextManager
from typing import Optional

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.exc import SQLAlchemyError, IntegrityError

# engine обычно создаётся один раз на приложение
engine = create_engine("postgresql+psycopg://user:pass@localhost:5432/app", pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

class UnitOfWork(AbstractContextManager):
    def __init__(self, session_factory=SessionLocal):
        self._session_factory = session_factory
        self.session: Optional[Session] = None

    def __enter__(self):
        self.session = self._session_factory()
        # Транзакция создаётся неявно SQLAlchemy-сессией.
        # Можно управлять явно через session.begin(), но ниже оставим читабельный вариант.
        return self

    def __exit__(self, exc_type, exc, tb):
        assert self.session is not None

        try:
            if exc_type is None:
                # Коммит — единая точка
                self.session.commit()
            else:
                # Роллбек — чтобы не оставить частичных изменений
                self.session.rollback()
        except SQLAlchemyError:
            # Если commit/rollback сами упали, это уже уровень "катастрофы".
            # Логи и re-raise — обычно правильная стратегия.
            self.session.rollback()
            raise
        finally:
            self.session.close()
        # Возвращение False означает: исключение пробрасывается наружу
        return False

Почему autoflush=False важно

С SQLAlchemy можно столкнуться с эффектом, когда «внешне ничего не коммитили», но при определённом обращении к данным SQLAlchemy сделает flush, а дальше вы получите ошибку ограничений внутри произвольного участка кода. Отключение autoflush иногда помогает держать контроль ближе к бизнес-логике: либо вы делаете flush осознанно, либо ошибка случается в ожидаемом месте (например, при коммите).

Если вам нужна стратегия с явным flush, можно встроить её в репозитории или UoW.

Применение UoW в бизнес-сервисе

Теперь покажем, как сервис использует UoW и репозитории.

Бизнес-операция: перевод

import uuid

class InsufficientFunds(Exception):
    pass

class TransferService:
    def __init__(self, uow_factory):
        self.uow_factory = uow_factory

    def transfer(self, *, from_account_id: int, to_account_id: int, amount: int):
        if amount <= 0:
            raise ValueError("amount must be positive")

        reference_id = str(uuid.uuid4())

        with self.uow_factory() as uow:
            accounts = AccountRepository(uow.session)
            ledger = LedgerRepository(uow.session)

            from_acc = accounts.get_by_id(from_account_id)
            if from_acc.balance < amount:
                raise InsufficientFunds("Not enough balance")

            accounts.decrement(from_account_id, amount)
            accounts.increment(to_account_id, amount)

            ledger.add_entry(
                account_id=from_account_id,
                amount=-amount,
                kind="TRANSFER_OUT",
                reference_id=reference_id,
            )
            ledger.add_entry(
                account_id=to_account_id,
                amount=amount,
                kind="TRANSFER_IN",
                reference_id=reference_id,
            )

            # Никаких commit — он в UoW.
            # При необходимости можно сделать flush здесь:
            # uow.session.flush()

Что важно:

• если на любом шаге возникнет исключение (включая ошибки БД, валидацию, ограничения), __exit__ сделает rollback;
• сервис не занимается транзакциями — только бизнес-логика;
• коммит гарантированно происходит один раз и только при успешном завершении.

Границы транзакции: когда начинать и когда заканчивать

Граница транзакции должна совпадать с границей бизнес-операции. То есть:

• открыли UoW — начинается транзакция;
• в рамках UoW выполняем все изменения;
• вышли из контекстного менеджера — завершили транзакцию (commit/rollback).

Типичная ошибка: “UoW на слишком широкий участок”

Если вы откроете UoW на весь запрос HTTP, включая долгие внешние вызовы (HTTP API, очереди, файловые операции), вы:

• держите транзакцию дольше, чем нужно;
• увеличиваете конкуренцию и вероятность блокировок;
• снижаете пропускную способность.

Решение: ограничивайте UoW ровно тем блоком кода, который реально должен быть атомарным относительно БД.

Обработка ошибок: что именно нужно делать

В нашем базовом UoW мы делаем роллбек по любому исключению. Это безопасно, но иногда полезно добавить различение ошибок:

• бизнес-ошибки (например, InsufficientFunds) — роллбек обязателен;
• ошибки уникальности / ссылочной целостности (IntegrityError) — тоже роллбек;
• временные сетевые проблемы — тоже роллбек, но иногда их нужно ретраить на уровне выше.

В SQLAlchemy важно, что после IntegrityError сессия может оказаться в состоянии, где дальнейшие операции бессмысленны без очистки. В простом варианте достаточно роллбека и проброса ошибки — UoW закроет сессию.

Если вы хотите более тонкую обработку, можно оборачивать commit в try/except и классифицировать ошибки.

Пример: классификация ошибок при commit

from sqlalchemy.exc import IntegrityError, SQLAlchemyError

class UnitOfWork(AbstractContextManager):
    # ...

    def __exit__(self, exc_type, exc, tb):
        assert self.session is not None

        try:
            if exc_type is None:
                try:
                    self.session.commit()
                except IntegrityError as e:
                    self.session.rollback()
                    raise  # или перевести в доменный Exception
            else:
                self.session.rollback()
        finally:
            self.session.close()
        return False

Явный begin() и контроль уровня изоляции (опционально)

Некоторые команды предпочитают явно управлять транзакцией через session.begin() — это делает намерение прозрачнее и снижает риск сюрпризов при настройках ORM.

Вариант UoW:

class UnitOfWork(AbstractContextManager):
    def __init__(self, session_factory=SessionLocal, isolation_level: str | None = None):
        self._session_factory = session_factory
        self._isolation_level = isolation_level
        self.session: Optional[Session] = None
        self._trans = None

    def __enter__(self):
        self.session = self._session_factory()
        if self._isolation_level:
            self._trans = self.session.begin()
            # Установка isolation_level зависит от драйвера/диалекта.
            # Обычно задаётся на стороне engine или через options.
        else:
            self._trans = self.session.begin()
        return self

    def __exit__(self, exc_type, exc, tb):
        assert self.session is not None

        try:
            if exc_type is None:
                self._trans.commit()
            else:
                self._trans.rollback()
        finally:
            self.session.close()
        return False

Если вы используете PostgreSQL и вам важно, например, сериализуемое чтение или борьба с аномалиями, параметры транзакции лучше обсуждать отдельно. Но в большинстве CRUD-сервисов достаточно стандартной схемы: один бизнес-сценарий = одна транзакция.

Тестирование без хрупких моков: две стратегии

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

Стратегия 1: интеграционные тесты с реальной БД (рекомендуется)

• Поднимаете тестовую БД (например, контейнер Docker).
• Прогоняете тесты против неё.
• Проверяете, что при ошибке транзакция откатывается: в таблицах не появляется частичных изменений.

Псевдокод/пример на pytest:

import pytest
from sqlalchemy.orm import Session

@pytest.fixture
def uow_factory(db_session_factory):
    # db_session_factory — фабрика для тестовой БД
    def _factory():
        return UnitOfWork(session_factory=db_session_factory)
    return _factory

def test_transfer_rolls_back_on_insufficient_funds(uow_factory):
    service = TransferService(uow_factory)

    # подготовка данных через отдельный session
    session: Session = next(iter([None]))  # замените на реальный получение session из фикстуры
    # В реальном тесте: вставьте изначальные записи: from_acc.balance < amount

    with pytest.raises(InsufficientFunds):
        service.transfer(from_account_id=1, to_account_id=2, amount=100)

    # Проверяем, что изменения не произошли:
    # - баланс from_acc не изменился
    # - баланс to_acc не изменился
    # - ledger entries не появились

Ключ — проверять состояние БД, а не «вызвали ли мы commit/rollback». Тогда тесты остаются устойчивыми к внутренним рефакторингам UoW.

Стратегия 2: тестовый double не для моков, а для контроля границ

Иногда вы хотите проверить, что сервис не делает commit напрямую. Но вместо моков репозиториев можно ввести тестовый session_factory, который выдаёт сессию-обёртку с логированием событий.

Например, можно создать EventTrackingSession (или прослушивать события SQLAlchemy), но без перехвата всех вызовов вручную. Однако это сложнее; интеграционный тест обычно проще и надёжнее.

Типичные ошибки при внедрении Unit of Work

1) Репозитории всё ещё делают commit

Это ломает главную гарантию UoW. Если вам кажется, что «коммит нужен, чтобы сохранить ID», некоммитный flush() решает проблему и не закрывает транзакцию.

2) UoW создаётся слишком часто или используется не по назначению

Например, сервис вызывает несколько подсервисов, и каждый подсервис открывает свой UoW. Итог — вместо одной атомарной операции вы получаете несколько транзакций.

Правило: один бизнес-сценарий — один UoW. Если вложенность нужна, UoW лучше передавать вниз как зависимость, а не создавать заново.

3) Долгие операции внутри транзакции

HTTP-вызовы, чтение больших файлов, ожидание очередей — всё это должно быть вынесено за границы UoW. Транзакция должна жить минимально возможное время.

4) Ошибки и состояние сессии

После некоторых ошибок (например, IntegrityError) сессия может быть в невалидном состоянии. В простом варианте UoW всё равно делает rollback и закрывает сессию — это нормальная и безопасная стратегия. Сложнее становится, если вы пытаетесь продолжать работу в той же сессии без пересоздания.

5) Нечёткие границы доменной логики и транзакционной логики

UoW — не место для бизнес-правил. Он про инфраструктуру транзакций и консистентность. Вся бизнес-логика должна оставаться в сервисах/доменных объектах.

Как связать UoW с Dependency Injection и “чистыми” зависимостями

В большинстве проектов UoW удобно передавать в сервис через фабрику:

• TransferService(uow_factory=...)
• в обработчике HTTP: service = TransferService(uow_factory)
• вызов: service.transfer(...)

Если вы используете контейнер DI (например, punq, dependency-injector, wired, etc.), UoW factory можно регистрировать как provider, который создаёт новый экземпляр на каждый запрос.

Суть: UoW должен быть request-scoped (в терминах web) или operation-scoped (в терминах консумера очереди/cron).

Практический чек-лист внедрения

1. Определите бизнес-сценарий: что должно быть атомарным?
2. Введите UnitOfWork как context manager.
3. Уберите commit из репозиториев.
4. Сервисная логика меняет сущности в рамках одного UoW.
5. Проверьте корректность на негативных кейсах:
   • недостаточно средств;
   • нарушение уникальности;
   • ошибка ссылочной целостности;
   • исключение в середине операции.
6. Покройте интеграционными тестами состояние БД после падений.
7. Следите за временем жизни транзакции: не держите её на внешних вызовах.

Итог: единая точка коммита как инженерная дисциплина

Unit of Work — не магия и не “ещё один паттерн ради паттерна”. Это способ дисциплинировать транзакционную часть системы: централизовать commit/rollback, изолировать репозитории от инфраструктуры и гарантировать, что бизнес-операции либо применяются полностью, либо не применяются вообще.

Если вы строите Python-сервис на SQLAlchemy (или похожей ORM/DAO-архитектуре), внедрение UoW обычно проходит без радикальных изменений доменной логики: репозитории перестают управлять транзакциями, сервисы получают контекст через единый объект, а тесты начинают проверять консистентность данных — не “вызвались ли методы”, а “что реально оказалось в БД”.

Если хочется глубже разобраться в стилях архитектуры (включая границы слоёв, варианты реализации UoW под async, тестовую стратегию и нюансы с SQLAlchemy), полезно посмотреть специализированный материал, например курс по теме здесь. Но даже без него, опираясь на приведённый скелет и чек-лист, вы сможете внедрить UoW в свой код так, чтобы атомарность и управляемость транзакций стали свойством системы, а не удачным совпадением.