6.Паттерн Unit of Work(Единица работы)
Паттерн Unit of Work(Единица работы)
В этой главе мы познакомимся с заключительной частью головоломки, которая связывает воедино паттерны уровня Репозитория и Уровня Сервиса: паттерн Unit of Work.
Если шаблон хранилища-это наша абстракция по отношению к идее постоянного хранения, то шаблон "Единицы Работы" (UoW) - это наша абстракция по отношению к идее атомарных операций. Это позволит нам окончательно и полностью отделить наш уровень обслуживания от уровня данных.
[before_uow_diagram] показывает, что в настоящее время много взаимодействий происходит между уровнями нашей инфраструктуры: API обращается напрямую к уровню базы данных, чтобы начать сеанс, он обращается к уровню репозитория для инициализации SQLAlchemyRepository, и он связывается с уровнем сервиса, чтобы выделить позиции.
 | Код для этой главы находится в ветке chapter_06_uow on GitHub: |
[after_uow_diagram] показывает наше целевое состояние. API Flask теперь делает только две вещи: он инициализирует единицу работы и вызывает службу. Сервис сотрудничает с UoW (нам нравится думать о UoW как о части сервисного уровня), но ни сама сервисная функция, ни Flask теперь не нуждаются в непосредственном общении с базой данных.
И мы сделаем все это с помощью прекрасного синтаксиса Python - диспетчера контекста.
Unit of Work взаимодействует с репозиторием
Давайте посмотрим, как работает единица работы (или UoW, что мы произносим как «you-wow»). Вот как будет выглядеть сервисный слой, когда мы закончим:
def allocate(
orderid: str, sku: str, qty: int,
uow: unit_of_work.AbstractUnitOfWork
) -> str:
line = OrderLine(orderid, sku, qty)
with uow: #
batches = uow.batches.list() #
...
batchref = model.allocate(line, batches)
uow.commit() #
| Мы запустим UoW в качестве менеджера контекста. |
| uow.batches это пакетный репо, поэтому UoW предоставляет нам доступ к нашему постоянному хранилищу. |
| Когда мы закончили, мы фиксируем или откатываем нашу работу, используя UoW. |
UoW действует как единственная точка входа в наше постоянное хранилище и отслеживает, какие объекты были загружены, и последнее состояние.[1]
Это дает нам три полезных преимущества:
Стабильный моментальный снимок базы данных для работы, поэтому объекты, которые мы используем, не меняются на полпути операции.
Способ сохранить все наши изменения сразу, чтобы, если что-то пойдет не так, мы не оказались в непоследовательном состоянии.
Простой API для наших проблем персистентности и удобное место для получения репозитория
Тест-драйв UoW с интеграционными тестами
Here are our integration tests for the UOW:
def test_uow_can_retrieve_a_batch_and_allocate_to_it(session_factory):
session = session_factory()
insert_batch(session, 'batch1', 'HIPSTER-WORKBENCH', 100, None)
session.commit()
uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory) #
with uow:
batch = uow.batches.get(reference='batch1') #
line = model.OrderLine('o1', 'HIPSTER-WORKBENCH', 10)
batch.allocate(line)
uow.commit() #
batchref = get_allocated_batch_ref(session, 'o1', 'HIPSTER-WORKBENCH')
assert batchref == 'batch1'
| Мы инициализируем UoW, используя нашу настраиваемую фабрику сеансов, и возвращаем объект uow для использования его в нашем блоке with. |
| UoW дает нам доступ к репозиторию batch через uow.batches. |
| Мы вызываем для него commit(), когда закончим. |
Для любопытных хелперы insert_batch и get_allocated_batch_ref выглядят так:
def insert_batch(session, ref, sku, qty, eta):
session.execute(
'INSERT INTO batches (reference, sku, _purchased_quantity, eta)'
' VALUES (:ref, :sku, :qty, :eta)',
dict(ref=ref, sku=sku, qty=qty, eta=eta)
)
def get_allocated_batch_ref(session, orderid, sku):
[[orderlineid]] = session.execute(
'SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku',
dict(orderid=orderid, sku=sku)
)
[[batchref]] = session.execute(
'SELECT b.reference FROM allocations JOIN batches AS b ON batch_id = b.id'
' WHERE orderline_id=:orderlineid',
dict(orderlineid=orderlineid)
)
return batchref
Unit of Work и её менеджер контекста
В наших тестах мы неявно определили интерфейс для того, что должен делать UoW. Давайте сделаем это явным с помощью абстрактного базового класса:
class AbstractUnitOfWork(abc.ABC):
batches: repository.AbstractRepository #
def __exit__(self, *args): #
self.rollback() #
@abc.abstractmethod
def commit(self): #
raise NotImplementedError
@abc.abstractmethod
def rollback(self): #
raise NotImplementedError
| UoW предоставляет атрибут под названием .batches, который дает нам доступ к репозиторию пакетов. |
| Если вы никогда не видели контекстного менеджера, __enter__ и __exit__ это два волшебных метода, которые выполняются, когда мы входим в блок with и когда выходим из него, соответственно. Это наши фазы setup и teardown. |
| Мы вызовем этот метод, чтобы явно зафиксировать нашу работу, когда будем готовы. |
| Если мы не фиксируем, или если мы выходим из диспетчера контекста, вызывая ошибку, мы выполняем «откат» rollback. (Откат не возымеет никакого эффекта, если была вызвана функция commit(). Читайте дальше для более подробного обсуждения этого вопроса.) |
Реальная Unit of Work Использует Сеансы SQLAlchemy
Главное, что добавляет наша конкретная реализация, - это сеанс базы данных:
DEFAULT_SESSION_FACTORY = sessionmaker(bind=create_engine( #
config.get_postgres_uri(),
))
class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
def __init__(self, session_factory=DEFAULT_SESSION_FACTORY):
self.session_factory = session_factory #
def __enter__(self):
self.session = self.session_factory() # type: Session #
self.batches = repository.SqlAlchemyRepository(self.session) #
return super().__enter__()
def __exit__(self, *args):
super().__exit__(*args)
self.session.close() #
def commit(self): #
self.session.commit()
def rollback(self): #
self.session.rollback()
| Модуль определяет фабрику сеансов по умолчанию, которая будет подключаться к Postgres, но мы позволяем переопределить это в наших интеграционных тестах, чтобы вместо этого мы могли использовать SQLite. |
| Метод __enter__ отвечает за запуск сеанса базы данных и создание экземпляра реального репозитория, который может использовать этот сеанс. |
| Закрываем сессию при выходе. |
| Наконец, мы предоставляем конкретные методы commit() и rollback(), которые используют наш сеанс базы данных. |
Иммитация Unit of Work для теста
Вот как мы используем фиктивный UoW в наших тестах уровня сервиса:
class FakeUnitOfWork(unit_of_work.AbstractUnitOfWork):
def __init__(self):
self.batches = FakeRepository([]) #
self.committed = False #
def commit(self):
self.committed = True #
def rollback(self):
pass
def test_add_batch():
uow = FakeUnitOfWork() #
services.add_batch("b1", "CRUNCHY-ARMCHAIR", 100, None, uow) #
assert uow.batches.get("b1") is not None
assert uow.committed
def test_allocate_returns_allocation():
uow = FakeUnitOfWork() #
services.add_batch("batch1", "COMPLICATED-LAMP", 100, None, uow) #
result = services.allocate("o1", "COMPLICATED-LAMP", 10, uow) #
assert result == "batch1"
...
| FakeUnitOfWork и FakeRepository тесно связаны, так же как реальные классы UnitofWork и Repository. Это прекрасно, потому что мы признаем, что объекты являются соавторами. |
| Обратите внимание на сходство с фальшивой функцией commit() из FakeSession (от которой теперь мы можем избавиться). Но это существенное улучшение, потому что мы сейчас подделываем код, который мы написали, а не сторонний код. Как гласит народная мудрость, "Не твоё — не трогай". |
| В наших тестах мы можем создать экземпляр UoW и передать его на наш уровень обслуживания, а не передавать репозиторий и сеанс. Это значительно изящнее. |
Использование UoW в сервисном слое
Вот как выглядит наш новый уровень обслуживания:
def add_batch(
ref: str, sku: str, qty: int, eta: Optional[date],
uow: unit_of_work.AbstractUnitOfWork #
):
with uow:
uow.batches.add(model.Batch(ref, sku, qty, eta))
uow.commit()
def allocate(
orderid: str, sku: str, qty: int,
uow: unit_of_work.AbstractUnitOfWork #
) -> str:
line = OrderLine(orderid, sku, qty)
with uow:
batches = uow.batches.list()
if not is_valid_sku(line.sku, batches):
raise InvalidSku(f'Invalid sku {line.sku}')
batchref = model.allocate(line, batches)
uow.commit()
return batchref
| Наш уровень обслуживания теперь имеет только одну зависимость, опять же от abstract UoW. |
Явные тесты для режима Commit/Rollback
Чтобы убедиться, что поведение commit/rollback фиксации/отката работает, мы написали несколько тестов:
def test_rolls_back_uncommitted_work_by_default(session_factory):
uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
with uow:
insert_batch(uow.session, 'batch1', 'MEDIUM-PLINTH', 100, None)
new_session = session_factory()
rows = list(new_session.execute('SELECT * FROM "batches"'))
assert rows == []
def test_rolls_back_on_error(session_factory):
class MyException(Exception):
pass
uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
with pytest.raises(MyException):
with uow:
insert_batch(uow.session, 'batch1', 'LARGE-FORK', 100, None)
raise MyException()
new_session = session_factory()
rows = list(new_session.execute('SELECT * FROM "batches"'))
assert rows == []
| Мы не показывали его здесь, но, возможно, стоит протестировать некоторые из более "неясных" действий базы данных, таких как транзакции, против "реальной" базы данных—то есть того же самого движка. На данный момент нам сходит с рук использование SQLite вместо Postgres, но в [chapter_07_aggregate] мы переключим некоторые тесты на использование реальной базы данных. Очень удобно, что наш класс UoW делает это легко! |
Явные и неявные коммиты
Теперь мы вкратце остановимся на различных способах реализации паттерна UoW.
Мы могли бы представить себе несколько иную версию UoW, которая фиксируется по умолчанию и откатывается только в том случае, если замечает исключение:
class AbstractUnitOfWork(abc.ABC):
def __enter__(self):
return self
def __exit__(self, exn_type, exn_value, traceback):
if exn_type is None:
self.commit() #
else:
self.rollback() #
| Должны ли мы иметь на счастливом пути неявную фиксацию? |
| И откатиться только при исключении? |
Это позволило бы нам сохранить строку кода и удалить явную фиксацию из нашего клиентского кода:
def add_batch(ref: str, sku: str, qty: int, eta: Optional[date], uow):
with uow:
uow.batches.add(model.Batch(ref, sku, qty, eta))
# uow.commit()
Это субъективное мнение, но мы, как правило, предпочитаем требовать явной фиксации, так что нам приходится выбирать, когда сбросить состояние.
Хотя мы используем дополнительную строку кода, это делает программное обеспечение безопасным по умолчанию. Поведение по умолчанию - "ничего не менять". В свою очередь, это делает наш код более простым для рассуждения, потому что есть только один путь кода, который ведет к изменениям в системе: полный успех и явная фиксация. Любой другой путь кода, любое исключение, любой ранний выход из области действия UoW приводит к безопасному состоянию.
Точно так же мы предпочитаем откат по умолчанию, потому что это легче понять; это откат к последней фиксации, поэтому пользователь либо выполнил задание, или мы сдуем их изменения. Сурово, но просто.
Примеры: Использование UoW для группировки нескольких операций в атомарную единицу
Ниже приведены некоторые примеры используемых схем работы. Это может привести к более простому рассуждению о том, как блоки кода работают совместно.
Пример 1: Перераспределение
Предположим, что мы хотим отменить распределение, а затем передислоцировать заказ:
def reallocate(line: OrderLine, uow: AbstractUnitOfWork) -> str:
with uow:
batch = uow.batches.get(sku=line.sku)
if batch is None:
raise InvalidSku(f'Invalid sku {line.sku}')
batch.deallocate(line) #
allocate(line) #
uow.commit()
| Если deallocate() не работает, очевидно мы не хотим вызывать allocate(). |
| Если allocate() терпит неудачу, вероятно мы, так же не хотим фиксить deallocate() |
Пример 2: Изменить размер партии
Наша судоходная компания звонит нам, чтобы сообщить, что одна из дверей контейнера открылась, и половина наших диванов упала в Индийский океан. Ой!
def change_batch_quantity(batchref: str, new_qty: int, uow: AbstractUnitOfWork):
with uow:
batch = uow.batches.get(reference=batchref)
batch.change_purchased_quantity(new_qty)
while batch.available_quantity < 0:
line = batch.deallocate_one() #
uow.commit()
| Здесь нам может понадобиться разобраться с любым количеством строк. Если мы получим неудачу на каком-то этапе, мы, вероятно, не захотим вносить никаких изменений. |
Уборка интеграционных тестов
Теперь у нас есть три набора тестов, все из которых, по сути, направлены на базу данных: test_orm.py, test_repository.py, и test_uow.py. Может, выкинем что-нибудь?
└── tests
├── conftest.py
├── e2e
│ └── test_api.py
├── integration
│ ├── test_orm.py
│ ├── test_repository.py
│ └── test_uow.py
├── pytest.ini
└── unit
├── test_allocate.py
├── test_batches.py
└── test_services.py
Вы всегда можете отказаться от тестов, если считаете, что они не принесут пользы в долгосрочной перспективе. Мы бы сказали, что test_orm.py был в первую очередь инструментом для изучения SQLAlchemy, поэтому в дальнейшем он не понадобится, особенно если основные вещи, которые он делает, описаны в test_repository.py. Этот последний тест вы можете оставить, но мы, безусловно, видим аргумент в пользу того, чтобы просто держать все на максимально возможном уровне абстракции (так же, как мы делали это в юнит-тестах).
| Это еще один пример урока из [chapter_05_high_gear_low_gear]: по мере того как мы строим лучшие абстракции, мы можем перемещать наши тесты, чтобы работать с ними, что оставляет нам свободу изменять лежащие в их основе детали. |
Заключение
Надеюсь, мы убедили вас, что шаблон «Единица работы» полезен и что диспетчер контекста - действительно хороший питонический способ визуальной группировки кода в блоки, которые мы хотим реализовать атомарно.
Этот шаблон настолько полезен, что SQLAlchemy уже использует UoW в форме объекта Session. Объект Session в SQLAlchemy - это способ, которым ваше приложение загружает данные из базы данных.
Каждый раз, когда вы загружаете новую сущность из базы данных, сеанс начинает отслеживать изменения в сущности, и когда сеанс сбрасывается, все ваши изменения сохраняются вместе. Зачем нам пытаться абстрагировать сеанс SQLAlchemy, если он уже реализует нужный нам паттерн?
[chapter_06_uow_tradeoffs] обсуждает некоторые компромиссы.
| Плюсы | Минусы |
|---|---|
|
|
Во-первых, Session API богат и поддерживает операции, которые нам не нужны или не нужны в нашем домене. Наш UnitOfWork упрощает сеанс до его основного ядра: его можно запустить, зафиксировать или выбросить.
Во-вторых, мы используем UnitOfWork для доступа к нашим объектам Repository. Это добавит удобства в использовании разработчиками, и это то, то мы не смогли бы сделать с помощью простого SQLAlchemy Session.
Наконец, мы снова мотивированы принципом инверсии зависимостей: наш уровень сервиса зависит от тонкой абстракции, и мы прикрепляем конкретную реализацию к внешнему краю системы. Это хорошо согласуется с собственной рекомендацией SQLAlchemy:
Держите жизненный цикл сеанса (и, как правило, транзакции) отдельным и внешним. Наиболее комплексный подход, рекомендуемый для более существенных приложений, будет стараться держать детали сеанса, транзакции и управления исключениями как можно дальше от деталей программы, выполняющей свою работу.
Комментарии
Отправить комментарий