/
woodpecker86
/
cosmic-python-code
Обзор
Документация
Войти
/
woodpecker86
/
cosmic-python-code
Код
Запросы
0
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/allocation/service_layer/unit_of_work.py
64 строки
2 KB
Harry
Uow no longer puts events directly on the bus [uow_collect_new_events]
01 мар 2024, 18:05
01 мар 2024, 18:05
a8ec47f
Код
Авторство
О чём код?
# pylint: disable=attribute-defined-outside-init from __future__ import annotations import abc from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.orm.session import Session from allocation import config from allocation.adapters import repository class AbstractUnitOfWork(abc.ABC): products: repository.AbstractRepository def __enter__(self) -> AbstractUnitOfWork: return self def __exit__(self, *args): self.rollback() def commit(self): self._commit() def collect_new_events(self): for product in self.products.seen: while product.events: yield product.events.pop(0) @abc.abstractmethod def _commit(self): raise NotImplementedError @abc.abstractmethod def rollback(self): raise NotImplementedError DEFAULT_SESSION_FACTORY = sessionmaker( bind=create_engine( config.get_postgres_uri(), isolation_level="REPEATABLE READ", ) ) 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.products = 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()