Architecture Patterns with Python(6장)

작업 단위 패턴(Unit of Work)

 

이번 장에서 작업 단위 패턴을 이용해 서비스 계층과 데이터 계층을 완전히 분리해본다.

 

기존에는 플라스크 앱이 직접 데이터베이스에 요청해 세션을 시작하고, 저장소 계층과 대화하여 SQLAlchemyRepository 를 초기화하며 서비스 계층에 할당을 요청한다.

 

UoW 가 있는 경우 플라스크 API 는 작업 단위를 초기화하고, 서비스를 호출하는 일만 하게 된다. 파이썬에서 콘텍스트 관리자로 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

 

- UoW 가 with 문과 함께 시작한다. 콘텍스트 관리자로 정의했기 때문이다.

- uow.batches 는 배치 저장소다. UoW가 데이터베이스에 대한 접근을 가능하게 한다.

- 작업이 완료되면 UoW 를 통해 커밋 혹은 롤백한다.

 

UoW 는 디비에 대한 단일 진입점으로 작용한다. 또한 어떤 객체가 메모리에 적재됐고 어떤 객체가 최종 상태인지를 기억한다.

 

UoW 에 대한 통합 테스트는 아래와 같이 할 수 있다.

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"

 

UoW 를 테스트할 때 필요한 도우미 함수 insert_batch 와 get_allocated_batch_ref 를 따로 정의하였다.

- 커스텀 세션 팩토리를 사용해 UoW 를 초기화하고 블록 안에서 사용할 uow 객체를 얻는다.

- uow.batches 를 통해 배치 저장소에 대한 접근을 제공한다.

- 작업이 끝나면 commit 한다.

 

그러면 이러한 일이 가능하게 하는 UoW 를 추상 클래스로 인터페이스를 명시해본다.

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

- UoW 는 .batches 라는 속성을 제공한다.

- __enter__ , __exit__ 는 with 블록에 들어갈 때와 나올 때 호출되는 매직 매서드다.

- 커밋을 하지 않거나 예외를 발생시켜서 콘텍스트 관리자를 빠져나가면 rollback 을 수행한다.

- commit() 이 이미 호출된 경우에는 롤백을 해도 아무 일이 발생하지 않는다.

 

위의 조건들을 만족하는 SQLAlchemy 세션을 사용하는 UoW 를 정의하면 아래와 같다.

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와 연결하는 default session factory 를 정의한다.

- __enter__ 메서드는 디비 세션을 시작하고 세션을 사용할 실제 저장소를 인스턴스화한다.

- 콘텍스트 관리자에서 나올 때 세션을 닫는다.

- 구체적인 commit() 과 rollback() 메서드를 제공한다.

 

이제 구현된 UoW 를 기반으로 서비스 계층 테스트에서 가짜 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"

 

커밋과 롤백이 제대로 작동하는지 확인하기 위한 테스트를 작성해 볼수도 있다.

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

 

이렇게 하면 commit() 을 명시하는 한 줄을 줄일 수 있다. 하지만 언제 상태를 디비에 반영할지(flush) 선택해야 한다면 명시적인 커밋을 요구하는 쪽이 더 선호된다.

명시적 커밋이 요구되면 소프트웨어 동작이 안전해진다. 또한 시스템의 상태를 바꾸는 경로가 단 하나(커밋을 명시하는 경로) 만 존재하므로 코드를 추론하기도 쉬워진다.

 

작업 단위 패턴(UoW) 을 통해서 트랜잭션 시작과 끝을 명시적으로 제어할 수 있다. 또한 원자적 연산을 표현하는 추상화이며, 콘텍스트 관리자를 사용하여 원자적으로 한 그룹으로 묶여야하는 코드 블록을 시각적으로 쉽게 알아볼 수 있다.

 

참고) 원자적 연산 : 수행되는 동안 어떠한 방해도 받지 않아야 하는 더 이상 분리할 수 없는 작업 단위를 말한다. 위의 예제에서는 디비와의 통신 과정에서 동시에 공유하고 있는 디비에 여러 접근이 이루어져, 데이터의 정합성에 영향을 줄 수 있는 상황을 방지할 수 있도록 원자적인 연산 행위에 대한 commit() 과 rollback() 을 콘텍스트 관리자로 구현해놓은 부분을 말한다.

 

반면, ORM 이 이미 좋은 추상화를 제공하는 경우가 있기 때문에 굳이 따로 UoW 를 구현하지 않아도 된다. 또한 다중 스레딩, 롤백 등 직접 구현할 때 상당히 신중해야 하는 케이스에 대해서는 ORM 이 제공하는 기능을 사용하는 것이 더 안전할 수 있다.

 

 

이미지 출처 : https://www.cosmicpython.com/book/chapter_06_uow.html