Привет, Хабр!

Unit of Work отслеживает все объекты, которые были загружены в память и изменены в ходе выполнения программы. Он управляет их состояниями и сохраняет изменения в базе данных в конце транзакции. Это делается с использованием сессий, которые действуют как контейнеры для всех изменений.

Когда работа завершена, Unit of Work выполняет commit для всех изменений, сохраняя их в базе данных. Если что-то пошло не так, выполняется rollback, и база данных возвращается в состояние до начала транзакции.

В данной статье рассмотрим, как реализовать паттерн Unit of Work с использованием SQLAlchemy.

Установим саму библиотеку:

pip install sqlalchemy

Основные компоненты Unit of Work в SQLAlchemy

Сессия — это главный компонент, через который проходят все взаимодействия с базой данных. Она отвечает за управление объектами, отслеживание изменений и выполнение транзакций. Можно сказать, что сессия — это основной рабочий инструмент, через который ORM управляет состоянием объектов и синхронизацией их с базой данных.

Сессия в SQLAlchemy создается с использованием фабрики сессий, обычно через sessionmaker, который связывает сессию с определенным engine — объектом, который управляет подключением к базе данных. Пример:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# создаем подключение к базе данных
engine = create_engine('postgresql://user:password@localhost/mydatabase')

# создаем фабрику сессий
Session = sessionmaker(bind=engine)

# создаем экземпляр сессии
session = Session()

Здесь engine управляет низкоуровневыми аспектами подключения к базе данных, а сессия действует как обертка, которая управляет объектами и транзакциями на более высоком уровне.

Одна из основных задач сессии — отслеживание изменений, которые происходят с объектами. SQLAlchemy автоматически отслеживает состояние объектов, которые были загружены в сессию, и любые изменения, сделанные с этими объектами. Например, изменить атрибут объекта, SQLAlchemy пометит этот объект как "измененный":

# изменение объекта
user = session.query(User).filter_by(name='Volodya').first()
user.nickname = 'Vladimir'

# сессия отслеживает это изменение

Сессия хранит список всех измененных объектов и применяет эти изменения к базе данных при выполнении транзакции.

SQLAlchemy использует транзакции для атомарности операций. Т.е все изменения, сделанные в рамках одной транзакции, либо успешно применяются к базе данных, либо полностью откатываются, если что-то пошло не так. Сессия в SQLAlchemy автоматически управляет транзакциями.

Коммит — это момент, когда все изменения, накопленные сессией, применяются к базе данных. Если в процессе коммита возникает ошибка, SQLAlchemy автоматически выполнит откат транзакции:

try:
    session.commit()  # попытка записать изменения в базу данных
except:
    session.rollback()  # в случае ошибки откат изменений
    raise
finally:
    session.close()  # закрываем сессию

Механизмы работы сессии

SQLAlchemy приджеривается некоторого механизма управления состоянием объектов в сессии:

1. Transient: Объект создан, но еще не привязан к сессии и не существует в базе данных.

2. Pending: Объект добавлен в сессию с помощью метода add(), но изменения еще не зафиксированы в базе данных.

3. Persistent: Объект связан с сессией и уже существует в базе данных.

4. Detached: Объект был связан с сессией, но сейчас отсоединен от нее (например, после закрытия сессии).

Рассмотрим, как эти состояния проявляются:

# Создаем новый объект User (состояние Transient)
new_user = User(name='Alice', fullname='Alice Wonderland', nickname='AliceW')

# Добавляем объект в сессию (состояние Pending)
session.add(new_user)

# Коммит изменений (объект становится Persistent)
session.commit()

# Теперь объект является Persistent, но если мы закроем сессию
session.close()

# Объект перейдет в состояние Detached

Интеграция Unit of Work с паттерном Repository

Зачем это?

Паттерн Repository предоставляет абстракцию для доступа к данным, позволяя отделить бизнес-логику от деталей доступа к базе данных. Он действует как хранилище объектов домена, предоставляя интерфейсы для операций CRUD.

Паттерн Unit of Work управляет транзакциями, гарантируя, что все операции с данными в рамках одной транзакции выполняются атомарно.

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

from typing import Generic, TypeVar, List

T = TypeVar('T')

class Repository(Generic[T]):
    def add(self, entity: T) -> None:
        raise NotImplementedError

    def remove(self, entity: T) -> None:
        raise NotImplementedError

    def get_by_id(self, id) -> T:
        raise NotImplementedError

    def list(self) -> List[T]:
        raise NotImplementedError

Этот интерфейс задает основу для конкретных репозиториев, которые будут работать с определенными типами сущностей.

Теперь создадим конкретную реализацию репозитория, используя SQLAlchemy. В этом репозитории будем управлять сессиями с помощью паттерна Unit of Work:

from sqlalchemy.orm import Session
from typing import List

class SQLAlchemyRepository(Repository[T]):
    def __init__(self, session: Session):
        self.session = session

    def add(self, entity: T) -> None:
        self.session.add(entity)

    def remove(self, entity: T) -> None:
        self.session.delete(entity)

    def get_by_id(self, id) -> T:
        return self.session.query(T).filter_by(id=id).one()

    def list(self) -> List[T]:
        return self.session.query(T).all()

SQLAlchemyRepository полагается на Session для выполнения всех операций с базой данных. Это позволяет репозиторию быть абстрактным по отношению к типу сущностей, с которыми он работает.

А теперь реализуем саму интеграцию с Unit of Work. Создадим класс, который будет управлять несколькими репозиториями и сессией одновременно:

from contextlib import contextmanager

class UnitOfWork:
    def __init__(self, session_factory):
        self.session_factory = session_factory
        self._session = None
        self._repositories = {}

    @contextmanager
    def start(self):
        self._session = self.session_factory()
        try:
            yield self
            self._session.commit()
        except:
            self._session.rollback()
            raise
        finally:
            self._session.close()

    def register_repository(self, entity_type, repository):
        self._repositories[entity_type] = repository(self._session)

    def get_repository(self, entity_type):
        return self._repositories[entity_type]

Класс UnitOfWork управляет сессией, создавая и закрывая её, а также управляет репозиториями.

Теперь посмотрим, как это будет использоваться в приложении:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)

uow = UnitOfWork(session_factory=Session)

# регистрируем репозиторий
uow.register_repository(User, SQLAlchemyRepository)

with uow.start() as uow_session:
    user_repo = uow_session.get_repository(User)
    new_user = User(name='John Doe', email='john@example.com')
    user_repo.add(new_user)

    # все изменения будут зафиксированы автоматически в конце блока

Здесь вся работа с базой данных проходит через репозиторий, который в свою очередь управляется Unit of Work.

Пример реализации паттерна Unit of Work

Начнем с создания моделей для базы данных. В нашем случае это будут Book, User и Rental:

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
import datetime

Base = declarative_base()

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author = Column(String)
    available_copies = Column(Integer)

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

class Rental(Base):
    __tablename__ = 'rentals'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    book_id = Column(Integer, ForeignKey('books.id'))
    rental_date = Column(DateTime, default=datetime.datetime.utcnow)
    return_date = Column(DateTime, nullable=True)

    user = relationship('User')
    book = relationship('Book')

Эти модели представляют структуру таблиц в базе данных. Пользователи могут брать книги напрокат, и для этого создается запись в таблице rentals.

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

from sqlalchemy.orm import Session

class BookRepository:
    def __init__(self, session: Session):
        self.session = session

    def add(self, book: Book):
        self.session.add(book)

    def get_by_id(self, book_id: int) -> Book:
        return self.session.query(Book).filter_by(id=book_id).first()

    def list_all(self):
        return self.session.query(Book).all()

class UserRepository:
    def __init__(self, session: Session):
        self.session = session

    def add(self, user: User):
        self.session.add(user)

    def get_by_id(self, user_id: int) -> User:
        return self.session.query(User).filter_by(id=user_id).first()

class RentalRepository:
    def __init__(self, session: Session):
        self.session = session

    def add(self, rental: Rental):
        self.session.add(rental)

    def get_active_rentals_by_user(self, user_id: int):
        return self.session.query(Rental).filter_by(user_id=user_id, return_date=None).all()

Каждый репозиторий инкапсулирует логику работы с конкретной сущностью. Например, BookRepository управляет объектами Book, а RentalRepository — арендными операциями.

Теперь создадим класс Unit of Work, который будет управлять сессией и репозиториями:

from contextlib import contextmanager

class UnitOfWork:
    def __init__(self, session_factory):
        self.session_factory = session_factory
        self._session = None

    @contextmanager
    def start(self):
        self._session = self.session_factory()
        try:
            yield self
            self._session.commit()
        except Exception as e:
            self._session.rollback()
            raise e
        finally:
            self._session.close()

    @property
    def books(self) -> BookRepository:
        return BookRepository(self._session)

    @property
    def users(self) -> UserRepository:
        return UserRepository(self._session)

    @property
    def rentals(self) -> RentalRepository:
        return RentalRepository(self._session)

UnitOfWork управляет сессией и предоставляет доступ к репозиториям. Вся работа с данными происходит в контексте одного блока with.

Теперь посмотрим, как можно использовать Unit of Work для выполнения операций, например, аренды книги юзером:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# настройка подключения к базе данных
engine = create_engine('sqlite:///library.db')
SessionFactory = sessionmaker(bind=engine)

# создаем Unit of Work
uow = UnitOfWork(session_factory=SessionFactory)

# аренда книги
def rent_book(user_id: int, book_id: int):
    with uow.start() as uow_session:
        user = uow_session.users.get_by_id(user_id)
        book = uow_session.books.get_by_id(book_id)
        
        if book.available_copies > 0:
            book.available_copies -= 1
            rental = Rental(user_id=user.id, book_id=book.id)
            uow_session.rentals.add(rental)
        else:
            raise Exception("No available copies of this book")

# пример использования
rent_book(1, 2)  # пользователь с ID 1 берет напрокат книгу с ID 2

Все операции происходят в рамках одного блока with, и если что-то пойдет не так (например, книга уже занята), изменения будут автоматически откатаны.

Системная аналитика: что важно и откуда начать? Обсудим на открытом уроке 22 августа, на котором:

• Проведем обзор ключевых навыков системного аналитика;

• Поделимся рекомендациями по началу карьеры и путям развития;

• Дадим практические советы и инструменты для повышения квалификации.

Записаться на урок можно по ссылке.