Я последовательно покажу пример решения задачи, чтобы объяснить причины, по которым я решаю, что пора прибегнуть к использованию данного паттерна.

Представим, что для нашего интернет-магазина у нас есть бизнес-правило - "Создание клиента". В рамках этого бизнес-правила нужно создать клиента, и привязать к нему личного менеджера, который нужен, для решения любых вопросов клиента.

Первым делом я пойду писать UseCase. На самом деле тесты... Опустим этот момент, так как для статьи их реализация, не имеет большого значения.

from dataclasses import dataclass
from enum import strEnum
  

@dataclass
class ClientDTO:
    fio: str
    age: int
    description: str
    mobile: str
    email: str


class CreatePartnerUseCase:
    """Создание клиента с привязкой менеджера."""
    
    def execute(self, manager_id: int, client_data: ClientDTO) -> int:
        pass

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

Первое, что приходит на ум, так это репозиторий клиентов. Он будет служить абстракцией над уровнем хранения данных, и будет отвечать за транзакционную целостность, ведь наша сущность может сохраняться/восстанавливаться из несколько таблиц базы данных(а может, мы ходим за ней по сети, или на диск, кто знает, абстракция?).

from abc import ABC, abstractmethod


class AbstractClientRepository(ABC):
    # Если нет конкретных причин делить метод save на create и update, 
    # то я предпочитаю скрывать эту логику внутри save. Считаю, что это более 
    # близкий интерфейс для коллекции данных.
    def save(self, client: ClientEntity) -> None:
        """Создание клиента."""

Репозиторию, как видно из аннотации, требуется ClientEntity. Для простоты используем dataclass.

from dataclasses import dataclass


# В данный момент жизненного цикла сущности, 
# на клиенте нет никакой логики кроме хранения состояния.
@dataclass
class ClientEntity:
    id: int | None
    fio: str
    age: int
    description: str
    mobile: str
    email: str

Хорошо, с репозиторием клиентов "разобрались". Ещё нам нужно создать связь, между клиентом и закреплённым за ним менеджером. Нужен ещё один репозиторий!

class AbstractManagerBindRepository(ABC):
    def save(self, manager_bind: ManagerBindEntity) -> None:
        """Создание привязки менеджера к клиенту."""

Ну и сама сущность привязки:

@dataclass
class ManagerBindEntiry:
    id: int | None
    manager_id: int 
    client_id: int
    # Далее могут быть поля, например, для указания приоритета. 
    # Это понадобится, когда одному клиенту будут привязаны несколько 
    # менеджеров, а нам нужно получить самого приоритетного из свободных.

Кажется, всё. Вернёмся к сценарию использования:

class CreatePartnerUseCase:
    """Создание клиента с привязкой менеджера."""
    def __init__(
        self,
        client_repository: AbstractClientRepository,
        manager_bind_repository: AbstractManagerBindRepository,
    )
        self._client_repository = client_repository
        self._manager_bind_repository = manager_bind_repository
    
    def execute(self, manager_id: int, client_data: ClientDTO) -> None:
        client = Client(
            fio=client_data.fio,
            age=client_data.age,
            description=client_data.description,
            phone=client_data.phone,
            email=client_data.email,
        )
        # Есть разные подходы, на кого возложить ответственность за создание 
        # идентификаторов. По возможности я предпочитаю отдавать эту 
        # ответственность репозиторию, так как в большинстве случаев, 
        # идентификатором сущностей является инкремент. Да, я понимаю что 
        # репозиторий мутирует состояние моей сущности, но я контролирую силу с 
        # которой репозиторий влияет на это состояние. Это не идеально с точки
        # зрения архитектуры. Это компромисс управления сложностью ПО, на 
        # который я иду осознанно.
        self._client_repository.save(client)
        manager_bind = ManagerBind(manager_id=manager_id, client_id=client.id)
        self._manager_bind_repository.save(manager_bind)

Кажется, что всё готово. Конечно, вы скажете, что у нас нет реализаций репозиториев, нет контроллера, тестов и т. п. Но нам это сейчас и не нужно. Эта статья про Unit of work , а не полноценный гайд по написанию кода ?.

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

1. Создать клиента

2. Привязать к нему менеджера

С другой стороны, что случится, если ClientRepository сохранит клиента, а ManagerBindRepository завершится ошибкой? Без должной обработки такого случая появится, неконсистентность данных. В базе данных появится Client, которому не привязан ни один Manager, а это нарушение логической транзакционности бизнес-правила.

Как решить возникшую проблему, если нам нужна возможность выполнения либо всех действия логической транзакции, либо ни одного? Нам нужна транзакционная целостность не только на уровне репозитория, но и на уровне логической транзакции сценария использования. В этом нам и поможет Unit of work.

Вот мы и добрались, до раскрытия темы. Спасибо, что дождались!

Unit of work или "единица работы" - Поведенческий паттерн, который позволяет определить логическую транзакцию. Эта транзакция нужна, для объединения нескольких "маленьких" операций в одну "большую"(Acid).

В книге "Шаблоны корпоративных приложений", Мартин Фаулер даёт такое уточнение - "Unit of Work содержит список объектов, охватываемых бизнес транзакцией, координирует запись изменений в базу данных и разрешает проблемы параллелизма."

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

Давайте посмотрим, как будет выглядеть код с использованием UoW .

class AbstractCreatePartnerUoW(ABC):
    @property
    @abstractmethod
    def client_repository(self) -> AbstractClientRepository:
        raise NotImplementedError()

    @property
    @abstractmethod
    def manager_bind_repository(self) -> AbstractManagerRepository:
        raise NotImplementedError()
      
    @abstractmethod
    def __enter__(self) -> AbstractCreatePartnerUoW:
        raise NotImplementedError() 

    @abstractmethod
    def __exit__(exc_type, exc_val, traceback) -> bool:
        raise NotImplementedError

class CreatePartnerUseCase:
    """Создание клиента с привязкой менеджера."""
    def __init__(self, unit_of_work: AbstractCreatePartnerUoW)
        self._unit_of_work = unit_of_work
    
    def execute(self, manager_id: int, client_data: ClientDTO) -> None:
        client = Client(
            fio=client_data.fio,
            age=client_data.age,
            description=client_data.description,
            phone=client_data.phone,
            email=client_data.email,
        )

        with self._unit_of_work as uow:
            uow.client_repository.save(client)
            manager_bind = ManagerBind(
                manager_id=manager_id, 
                client_id=client.id,
            )
            uow.manager_bind_repository.save(manager_bind)

Все операции, выполненные репозиториями, в рамках UoW будут выполнены вместе. Если в любой момент работы UoW произойдёт ошибка, то UoW перехватит её в методе __exit__ и сделает rollback всей транзакции.

Как бы могла выглядеть реализация AbstractCreatePartnerUoW :

class DjangoUnitOfWork(AbstractUnitOfWork):
    def __init__(self):
        self.client_repository = DjangoClientRepository()
        self.manager_bind_repository = DjangoManagerRepository()
        
    def __enter__(self) -> DjangoUnitOfWork:
        transaction.set_autocommit(False)
        return self

    def __exit__(self, exc_type, exc_val, traceback) -> bool:
        if exc_type:
            transaction.rollback()
            return False
        self.commit()
        transaction.set_autocommit(True)

1. При начале работы __enter__ происходит отключение автоматического сохранения изменений.

2. После работы в контексте UoW или возникшей ошибки, управление попадает в __exit__ .

3. Происходит проверка, нужно ли откатить или сохранить изменения.

4. Возвращается автоматическое сохранение изменений.

Мартин Фаулер описывает этот паттерн, как паттерн, применимый именно к базам данных.

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

Некоторые люди отдают предпочтение UoW на уровне - "Нужно применять везде, где есть работа с репозиториями", что, как по мне, только удорожит разработку и сложность проекта.

Кто-то неправильно управляя сложностью ПО, может решить повесить @atomic прямо на CreatePartnerUseCase. Тем самым связав свою бизнес-логику с инфраструктурой и зависимостью от фреймворка.

Я же считаю, что место UoW именно там, где происходит изменение данных более одного раза за сценарий использования. В случаях, когда это происходит один раз за сценарий, как минимум это множество CRUD сервисов, которым априори не нужно делать несколько операций над хранилищем данных.

Так же есть мнение, что как только появляется UoW, репозитории должны "разучиться" сами следить за атомарностью своих операций и предоставить ответственность за это UoW.

Я не согласен с ними...

Используя паттерн репозиторий, я ожидаю, что он сам будет следить за атомарностью работы своих методов. Я выбираю этот паттерн, потому что он даёт мне возможность абстрагироваться над уровнем хранения данных. Я доверяю ему выполнение единичной операции. Если репозиторий не способен сохранить сущность либо полностью либо не сохранить её совсем(помним, что сущность может состоять из нескольких таблиц БД и наоборот.), то мне не нужен такой репозиторий.

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

Подытожим. Патерн Unit of work это хорошее решение, если вы хотите соблюсти логическую транзакционность внутри ваших сценариев использования.

Я показал достаточно простой вариант использования паттерна. При большом желании можно контролировать, какие именно операции произошли в рамках транзакции, и, например, отправлять событие в систему. Это может быть полезно, если вы используете паттерн наблюдатель или событийно ориентированную архитектуру.

Спасибо за внимание!