Что такое unit of work?

Unit of Work (Единица работы)

Unit of Work — это архитектурный паттерн, который отслеживает все объекты, изменённые во время бизнес-транзакции, и гарантирует, что все эти изменения сохраняются в базу данных одной атомарной операцией. Это одна из фундаментальных концепций в DDD (Domain-Driven Design) и современных ORMs.

Назначение Unit of Work

Паттерн решает следующие задачи:

1. Отслеживание изменений — система помнит, какие объекты были изменены
2. Атомарность — все изменения сохраняются вместе или откатываются вместе
3. Консистентность — гарантирует, что база данных остаётся в согласованном состоянии
4. Оптимизация — минимизирует количество SQL запросов

Как это работает

1. Начало транзакции (начало Unit of Work)
2. Загрузка объектов из БД
3. Изменение объектов в памяти
4. Отслеживание: система записывает, какие объекты изменились
5. Commit: система генерирует SQL-запросы и выполняет их
6. Если ошибка — Rollback: откат всех изменений
7. Конец транзакции (конец Unit of Work)

Пример без Unit of Work (плохо)

# ПЛОХО: Ручное управление каждым изменением
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

engine = create_engine("sqlite:///:memory:")
session = Session(engine)

# Создаём пользователя
user = User(name="John", email="john@example.com")
session.add(user)  # Добавляем в session
session.commit()  # Сохраняем

# Изменяем пользователя
user.email = "newemail@example.com"
session.commit()  # Снова сохраняем

# Удаляем пользователя
session.delete(user)
session.commit()  # И снова сохраняем

# Проблема: три отдельных commit'а, возможны проблемы при ошибках

Пример с Unit of Work (хорошо)

from sqlalchemy.orm import Session
from contextlib import contextmanager

class UnitOfWork:
    """Реализация паттерна Unit of Work для управления транзакциями"""
    
    def __init__(self, session: Session):
        self.session = session
    
    def __enter__(self):
        """Начало транзакции"""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Конец транзакции"""
        if exc_type is not None:
            self.rollback()
        else:
            self.commit()
    
    def commit(self):
        """Сохранить все изменения"""
        try:
            self.session.commit()
        except Exception as e:
            self.session.rollback()
            raise
    
    def rollback(self):
        """Откатить все изменения"""
        self.session.rollback()

# Использование с context manager
with UnitOfWork(session) as uow:
    user = User(name="John", email="john@example.com")
    uow.session.add(user)
    
    user.email = "newemail@example.com"
    uow.session.add(user)
    
    # Если ошибка — всё откатится автоматически
    # Если успех — всё сохранится в одной транзакции

Unit of Work в SQLAlchemy (встроенная реализация)

from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(bind=engine)

def process_user_registration(user_data: dict):
    """Unit of Work: создание пользователя и отправка письма"""
    session = SessionLocal()
    
    try:
        # Создаём пользователя
        user = User(
            name=user_data["name"],
            email=user_data["email"]
        )
        session.add(user)  # Отслеживание: user в session.new
        session.flush()  # Генерируем SQL, но не коммитим
        
        # Создаём запись в логах
        log = UserLog(user_id=user.id, action="registered")
        session.add(log)
        
        # Одна атомарная операция
        session.commit()
        
        return user
    except Exception:
        session.rollback()  # Откат всех изменений
        raise
    finally:
        session.close()

Состояния объектов в Session

# Transient (новый объект, не в session)
user = User(name="John")

# Persistent (в session, синхронизирован с БД)
session.add(user)
session.commit()

# Detached (был в session, но session закрыта)
session.close()
# Теперь user — detached

# Deleted (удалён)
session.delete(user)
session.commit()

Unit of Work с Repositories

from abc import ABC, abstractmethod

class BaseRepository(ABC):
    def __init__(self, session: Session):
        self.session = session
    
    @abstractmethod
    def add(self, entity):
        pass
    
    @abstractmethod
    def remove(self, entity):
        pass

class UserRepository(BaseRepository):
    def add(self, user: User):
        self.session.add(user)
    
    def remove(self, user: User):
        self.session.delete(user)
    
    def get(self, user_id: int) -> User:
        return self.session.query(User).filter(User.id == user_id).first()

# Unit of Work с Repository Pattern
class ServiceLayer:
    def __init__(self, session: Session):
        self.users = UserRepository(session)
        self.session = session
    
    def register_user(self, name: str, email: str):
        user = User(name=name, email=email)
        self.users.add(user)
        self.session.commit()
        return user

Unit of Work в FastAPI

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users")
def create_user(user_data: dict, db: Session = Depends(get_db)):
    """
    FastAPI автоматически управляет сессией и Unit of Work:
    - При успехе: commit
    - При ошибке: rollback
    """
    user = User(**user_data)
    db.add(user)  # Отслеживание
    db.commit()   # Сохранение
    return user

Преимущества Unit of Work

• Атомарность — все изменения сохраняются вместе
• Безопасность — откат при ошибках
• Производительность — минимизация SQL запросов
• Консистентность — база данных остаётся в согласованном состоянии
• Упрощение кода — не нужно писать try/except для каждой операции

Когда использовать

• Сложные операции с несколькими таблицами
• Когда нужна гарантия атомарности
• Микросервисная архитектура
• Когда используются DDD и Repository Pattern

Важно: Unit of Work входит в состав почти всех современных ORMs (SQLAlchemy, Django ORM) и часто используется автоматически, но понимание его принципов критично для написания надёжного кода.