[파이썬 아키텍처 패턴] 6장

class UnitOfWork: def __init__(self, session_factory): self.session_factory = session_factory self.session = None def __enter__(self): self.session = self.session_factory() return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None: self.rollback() else: self.commit() self.session.close() def commit(self): self.session.commit() def rollback(self): self.session.rollback() @property def batches(self): # 배치 저장소를 반환하는 메서드 pass

class BatchRepository: def __init__(self, session): self.session = session def add(self, batch): self.session.add(batch) def get(self, batch_id): return self.session.query(Batch).filter_by(id=batch_id).one()

def insert_batch(session, ref, sku, qty, eta): session.execute( 'INSERT INTO batches (reference, sku, qty, eta) VALUES (:ref, :sku, :qty, :eta)', dict(ref=ref, sku=sku, qty=qty, eta=eta) ) def get_allocated_batch_ref(session, orderid, sku): result = session.execute( 'SELECT b.reference FROM allocations a JOIN batches b ON a.batch_id = b.id JOIN order_lines l ON a.orderline_id = l.id WHERE l.orderid = :orderid AND l.sku = :sku', dict(orderid=orderid, sku=sku) ) return result.fetchone()[0]

def test_uow_can_retrieve_a_batch_and_allocate_to_it(sqlite_session_factory): session = sqlite_session_factory() insert_batch(session, 'batch1', 'sku1', 100, None) session.commit() uow = UnitOfWork(session_factory=sqlite_session_factory) with uow: batch = uow.batches.get('batch1') batch.allocate(OrderLine('order1', 'sku1', 10)) uow.commit() batch_ref = get_allocated_batch_ref(session, 'order1', 'sku1') assert batch_ref == 'batch1'

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

import pytest from allocation.domain import model from allocation.service_layer import unit_of_work 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 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"

from typing import TypeVar, Generic T = TypeVar('T', bound=AbstractRepository) class AbstractUnitOfWork(abc.ABC): batches: repository.AbstractRepository def __enter__(self) -> AbstractUnitOfWork: return self def __exit__(self, *args): self.rollback() @abc.abstractmethod def commit(self): raise NotImplementedError @abc.abstractmethod def rollback(self): raise NotImplementedError @property def batches(self) -> T: return self._batches

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()

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"

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 == []

class AbstractUnitOfWork(abc.ABC): batches: repository.AbstractRepository def __enter__(self) -> AbstractUnitOfWork: return self def __exit__(self, exn_type, exn_value, traceback): if exn_type is None: self.commit() else: self.rollback()