Паттерн Unit of Work в Python с SQLAlchemy

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

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 августа, на котором:

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

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

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

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