06. Unit of Work Pattern¶

Introduction¶

이번 장에서는 서비스 레이어를 데이터 레이어에서 완전히 분리하기 위한 레포지터리 패턴과 서비스 레이어 패턴을 통합하는 “Unit of Work 패턴”을 소개합니다.

지난 내용 복습¶

2장에서는 레포지터리 패턴으로 영구 저장소의 직접적인 구현으로부터 데이터 레이어를 분리했습니다.

4장에서는 서비스 레이어 패턴을 도입하여 여러 컴포넌트를 중간에서 조정하는(ochestration) 작업들을 “서비스 레이어”로 분리해냈습니다. (핵심 비즈니스 로직 – 도메인 서비스 – 는 여전히 도메인 모델에 남아있습니다.)

특히 3, 4장에서는 건전한 테스트 피라미드를 유지하기 위해 테스트를 개선하는 과정에서 설계가 거꾸로 개선되는(Test-induced design damage) 흥미로운 과정을 경험했습니다. E2E 테스트가 온갖 상황을 다 고려하지 않도록 개선하기 위해 엔드포인트는 웹 요청 처리 작업만 남겨두고 오케스트레이션 로직을 서비스 레이어로 분리한 것입니다. 이를 통해 오케스트레이션 작업의 테스트를 단위 테스트로 경량화 시킬 수있었습니다.

4장의 마지막 부분을 기억하신다면, 서비스 함수의 예를 들면서, 서비스 계층이 DB와 강한 결합을 맺고 있는 문제가 지적되었던 것이 떠오를 것입니다.

services/batch.py

class InvalidSku(Exception):
    pass

...

def allocate(line: OrderLine, repo: AbstractRepository, session) -> str:
    batches = repo.list()
    if not is_valid_sku(line.sku, batches):
        raise InvalidSku(f'Invalid sku {line.sku}')
    batchref = model.allocate(line, batches)
    session.commit()  # 세션 커
    return batchref

서비스가 직접 레포지터리나 DB세션을 조작하는 대신 이런 작업들을 추상화 한 UoW 서비스를 두어 이 서비스를 통해 레포지터리에 대한 작업이 수행되도록 할 것입니다.

기존 아키텍처에서는 Flask Endpoint가 서비스 API를 통해 직접 레포지터리와 DB에 접근했다면 UoW 패턴을 이용한 아키텍처에서는 서비스 API가 레포지터리 접근을 위해 별도로 분리된 UoW 서비스를 초기화 하여 이 서비스를 통해 DB에 접근하게 되는 것입니다.

2장부터 6장까지 개선된 아키텍처를 한 눈에 확인해 봅시다.

Sanity Checks for the project¶

Project structure¶

[1]:

!tree app -I '__pycache__' --sort=version

app
├── __init__.py
├── __main__.py
├── adapters
│   ├── __init__.py
│   ├── orm.py
│   └── repository.py
├── apps
│   ├── __init__.py
│   ├── fastapi.py
│   └── flask.py
├── config.py
├── domain
│   ├── __init__.py
│   └── models.py
├── requirements.txt
├── routes
│   ├── __init__.py
│   └── flask.py
└── services
    ├── __init__.py
    └── batch.py

5 directories, 16 files

[2]:

!mypy -p app --strict

Success: no issues found in 15 source files

[3]:

!pylint app


--------------------------------------------------------------------
Your code has been rated at 10.00/10 (previous run: 10.00/10, +0.00)

[4]:

!pytest

============================= test session starts ==============================
platform linux -- Python 3.9.1, pytest-6.1.2, py-1.10.0, pluggy-0.13.1
rootdir: /home/ykkim/notebooks/2021-msa-study/06-unit-of-work-pattern
plugins: flask-1.1.0, anyio-2.0.2
collected 27 items

tests/e2e/test_api.py .                                                  [  3%]
tests/integration/test_orm.py ......                                     [ 25%]
tests/integration/test_repository.py ..                                  [ 33%]
tests/unit/test_allocate.py ....                                         [ 48%]
tests/unit/test_batch.py .........                                       [ 81%]
tests/unit/test_services.py .....                                        [100%]

============================== 27 passed in 0.79s ==============================

테스트 피라미드는?¶

[5]:

from tests.utils import get_test_counts, show_test_pyramid
test_counts = get_test_counts(); test_counts

[5]:

{'unit': 18, 'integration': 8, 'e2e': 1}

[6]:

show_test_pyramid(test_counts)

[6]:

The Unit of Work Collaborates with the Repository¶

그럼 본격적으로 UoW(Unit of Work: 유-와우 라고 발음합니다)를 설계해봅시다.

[7]:

from unittest.mock import MagicMock
unit_of_work = MagicMock() # Mock 객체로 이런 서비스가 있다고 가정합니다.

[8]:

# services/batch.py

from app.domain import models

def allocate(
        orderid: str, sku: str, qty: int,
        uow: unit_of_work.AbstractUnitOfWork
) -> str:
    line = OrderLine(orderid, sku, qty)
    with uow: # ContextManager객체 - with 블록이 끝나면 레포지터리 리소스에 대한 정리 작업 수행
        batches = uow.batches.list()  # uow.batches 처럼 여러 레포지터리와 연결됨.
        ...
        batchref = models.allocate(line, batches)
        uow.commit()

uow를 ContextManager 타입의 객체로 구현하여 with 블록을 빠져나갈 때 자동으로 트랜잭선이 커밋/롤백 되고 리소스가 반환되도록 할 것입니다.

UoW 는 영구 저장소의 유일한 진입점이며, 로드된 객체의 최신 상태를 계속 트래킹 합니다. 이를 통해 얻을 수 있는 3가지 이득은 다음과 같습니다.

A stable snapshot of the database to work with, so the objects we use aren’t changing halfway through an operation
A way to persist all of our changes at once, so if something goes wrong, we don’t end up in an inconsistent state
A simple API to our persistence concerns and a handy place to get a repository

Test-Driving a UoW with Integration Tests¶

이제 통합 테스트를 성공시키며 실제 UoW 를 구현해보도록 합시다.

일단 UoW를 사용하여 배치를 조회하고 할당하는 가장 간단한 유즈케이스를 생각해보세요.

DB 세션을 얻고 세션을 이용해 저장소에 샘플 배치 하나를 추가 한뒤 커밋합니다.
세션을 이용해 UoW 서비스 객체를 생성합니다.
UoW API(uow.batches.get)를 이용해 레퍼런스에 대한 배치를 조회합니다.
Batch API(Batch.allocate)를 이용해 샘플 OrderLine을 이 배치에 할당합니다.
UoW API로 변경사항을 커밋한 뒤 UoW 서비스 사용을 완료하고 리소스를 반환합니다.
마지막으로, 저장소에 샘플 배치가 할당된 상태로 변경되었는지 확인합니다.

이 과정을 다음 코드로 기계적으로 옮길 수 있습니다. 실패하는 테스트를 만드는 것이 목적임에 유의하세요. unit_of_work 나 insert_batch, get_allocated_batch_ref 의 구현은 나중에 고치면 됩니다.

[9]:

from unittest.mock import MagicMock
from app.adapters.orm import init_engine, start_mappers, sessionmaker
from app import config
from tests import mytest

@mytest.fixture
def get_session():
    metadata = start_mappers()
    engine = init_engine(metadata, 'sqlite://') # using memory db
    return sessionmaker(engine)

get_allocated_batch_ref = MagicMock() # 함수가 있다고 가정 합니다.
insert_batch = MagicMock()            # 함수가 있다고 가정 합니다.

[tests/integration/test_uow.py]

[10]:

@mytest.test
def test_uow_can_retrieve_a_batch_and_allocate_to_it(get_session):
    session = get_session()
    insert_batch(session, 'batch1', 'HIPSTER-WORKBENCH', 100, None)
    session.commit()
    uow = unit_of_work.SqlAlchemyUnitOfWork(get_session)
    with uow:
        batch = uow.batches.get(reference='batch1')
        line = models.OrderLine('o1', 'HIPSTER-WORKBENCH', 10)
        batch.allocate(line)
        uow.commit()

    batchref = get_allocated_batch_ref(session, 'o1', 'HIPSTER-WORKBENCH')
    assert batchref == 'batch1'

❌ test_uow_can_retrieve_a_batch_and_allocate_to_it

Traceback (most recent call last):
  File "<ipython-input-10-f1a2a260e2a0>", line 14, in test_uow_can_retrieve_a_batch_and_allocate_to_it
    assert batchref == 'batch1'
AssertionError

insert_batch 와 get_allocated_batch_ref 는 다음과 같이 작성할 수 있을 것입니다.

[11]:

def insert_batch(session, ref, sku, qty, eta):
    session.execute(
        'INSERT INTO batch (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_line WHERE orderid=:orderid AND sku=:sku',
        dict(orderid=orderid, sku=sku)
    )
    [batchref, *_] = session.execute(
        'SELECT b.reference FROM allocation JOIN batches AS b ON batch_id = b.id'
        ' WHERE orderline_id=:orderlineid',
        dict(orderlineid=orderlineid)
    )
    return batchref

테스트를 다시 실행해보면 여전히 실패합니다.

[12]:

mytest.run(test_uow_can_retrieve_a_batch_and_allocate_to_it)

❌ test_uow_can_retrieve_a_batch_and_allocate_to_it

Traceback (most recent call last):
  File "<ipython-input-11-2e7452be49be>", line 10, in get_allocated_batch_ref
    [orderlineid, *_] = session.execute(
ValueError: not enough values to unpack (expected at least 1, got 0)

당연합니다. unit_of_work 서비스가 `MagicMock <https://docs.python.org/ko/3/library/unittest.mock.html#unittest.mock.MagicMock>`__ 객체였기 때문입니다. Mock 객체에 대한 작업은 아무 효과 없이 성공 처리 됩니다. 단 작업 과정이 Mock 객체에 다 기록됩니다. (때문에 Spy 객체라고도 볼 수 있습니다.)

Mock 사용법¶

다음은 Mock 객체 사용법 설명을 위한 예시 코드입니다.

[13]:

unit_of_work = MagicMock()
uow = unit_of_work.SqlAlchemyUnitOfWork(get_session())
uow.batches.get(reference='batch1')
uow.commit()
uow.commit()

[13]:

<MagicMock name='mock.SqlAlchemyUnitOfWork().commit()' id='140691514753904'>

[14]:

uow.batches.get.called, uow.batches.get.call_args

[14]:

(True, call(reference='batch1'))

[15]:

uow.commit.call_count

[15]:

이제 본격적으로 UoW 를 구현해봅시다.

Unit of Work and Its Context Manager¶

테스트케이스에서 우리는 UoW API를 이미 설계했습니다. UoW 객체에 다음과 같은 속성이 있다고 가정했습니다.

uow.batches: repository.AbstractRepository # Batch 모델 레포지터리
uow.commit() # 도메인 변경사항을 레포지터리에 커밋하는 함수

이를 좀 더 일반화 시켜 rollback 메소드 포함시키고 ContextManager를 이용한 자동 리소스 반환을 지원하는 아래와 같은 추상 클래스를 만들 수 있습니다.

[16]:

from contextlib import AbstractContextManager
import abc
from app.adapters import repository

class AbstractUnitOfWork(AbstractContextManager):
    batches: repository.AbstractRepository

    def __enter__(self):
        return self

    def __exit__(self, *args):
        """with 블록에서 빠져나갈 때 실행되는 메소드입니다."""
        self.rollback() # commit() 안되었을때 변경을 롤백합니다.
                        # (이미 커밋 되었을 경우 rollback은 아무 효과도 없음)

    @abc.abstractmethod
    def commit(self):
        raise NotImplementedError

    @abc.abstractmethod
    def rollback(self):
        raise NotImplementedError

추상 클래스로 정의되었기 때문에 이를 구체화하는 자식 클래스는 반드시 commit 과 rollback 메소드를 구현해야 합니다.

[17]:

@mytest.test
def test_uow_wrong_concrete_class():

    class TestUoW(AbstractUnitOfWork):
        pass

    uow = TestUoW()

❌ test_uow_wrong_concrete_class

Traceback (most recent call last):
  File "<ipython-input-17-0e4313d2cc3c>", line 7, in test_uow_wrong_concrete_class
    uow = TestUoW()
TypeError: Can't instantiate abstract class TestUoW with abstract methods commit, rollback

또한 ContextManager 를 지원하므로 with 블록과 같이 사용하면 블록이 끝났을 때 rollback 작업이 수행됩니다.

[18]:

@mytest.test
def test_uow_with_statement():

    class DummyUoW(AbstractUnitOfWork):
        def commit(self):
            print('call commit()!')

        def rollback(self):
            print('call rollback()!')

    with DummyUoW() as uow:
        pass

call rollback()!
✅ test_uow_with_statement

The Real Unit of Work Uses SQLAlchemy Sessions¶

AbstractUnitOfWork 를 상속받아 SQLAlchemy ORM 세션을 이용한 실제 UoW 를 구현해보겠습니다.

4장 실습에서 사용된 docker-compose.yml 과 동일한 설정으로 PostgreSQL DB가 떠 있다고 가정하겠습니다.

docker 컨테이너를 띄운 호스트에서 다음과 같이 확인해보세요.

$ docker-compose --project-directory ../04-flask-api-and-service-layer ps

   Name                Command              State              Ports
--------------------------------------------------------------------------------
microlab     /bin/sh -c /root/start.sh      Up      0.0.0.0:12222->22/tcp,
                                                    0.0.0.0:5005->80/tcp,
                                                    0.0.0.0:18888->8888/tcp
pgsql-test   docker-entrypoint.sh           Up      0.0.0.0:5432->5432/tcp
             postgres

목업을 준비하고 실제 UoW 객체를 작성해봅시다.

[20]:

config = MagicMock()
config.get_db_url = lambda: 'postgres://smprc:test@pgsql-test'

[services/unit_of_work.py]

[21]:

...
metadata = start_mappers()
engine = init_engine(metadata, config.get_db_url(), drop_all=True)
DEFAULT_SESSION_FACTORY = sessionmaker(engine)

class SqlAlchemyUnitOfWork(AbstractUnitOfWork):

    def __init__(self, get_session=DEFAULT_SESSION_FACTORY):
        self.get_session = get_session
        self.committed = False

    def __enter__(self):
        self.session = self.get_session() # 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.committed = True
        self.session.commit()

    def rollback(self):
        if self.committed:
            print('session rolled back', file=sys.stderr)
        self.session.rollback()

그럼 이 UoW 구현으로 실제로 DB와 연결해 테스트를 해보겠습니다.

[22]:

from app.domain.models import Batch

@mytest.test
def test_uow_with_real_session():

    with SqlAlchemyUnitOfWork() as uow:
        uow.batches.clear()
        uow.batches.add(Batch('b1', 'TEST', 10))
        uow.commit()

    with SqlAlchemyUnitOfWork() as uow:
        [batch, *_] = uow.batches.list()
        assert 'b1' == batch.reference
        uow.batches.delete(batch)

✅ test_uow_with_real_session

session rolled back

Fake Unit of Work for Testing¶

앞서 만든 UoW 구현은 실제 DB와 연결이 필요하기 때문에 단위 테스트용으로 사용할 수 없습니다.

단위 테스트에 사용 가능한 Fake 객체를 만들어 보겠습니다.

[23]:

from tests.unit.test_services import FakeRepository

[tests/unit/test_services.py]

[24]:

@mytest.fixture
def services():
    from app import services as services_ch5
    return services_ch5

class FakeUnitOfWork(AbstractUnitOfWork):

    def __init__(self):
        self.batches = FakeRepository([])
        self.committed = False

    def commit(self):
        self.committed = True

    def rollback(self):
        pass

@mytest.test
def test_add_batch(services):
    uow = FakeUnitOfWork()
    services.batch.add("b1", "CRUNCHY-ARMCHAIR", 100, None, uow)
    assert uow.batches.get("b1") is not None
    assert uow.committed

@mytest.test
def test_allocate_returns_allocation(services):
    uow = FakeUnitOfWork()
    services.batch.add("batch1", "COMPLICATED-LAMP", 100, None, uow)
    result = services.allocate("o1", "COMPLICATED-LAMP", 10, uow)
    assert "batch1", result

❌ test_add_batch

Traceback (most recent call last):
  File "<ipython-input-24-48c366eb98d3>", line 21, in test_add_batch
    services.batch.add("b1", "CRUNCHY-ARMCHAIR", 100, None, uow)
TypeError: add() missing 1 required positional argument: 'session'

❌ test_allocate_returns_allocation

Traceback (most recent call last):
  File "<ipython-input-24-48c366eb98d3>", line 28, in test_allocate_returns_allocation
    services.batch.add("batch1", "COMPLICATED-LAMP", 100, None, uow)
TypeError: add() missing 1 required positional argument: 'session'

테스트가 실패합니다. 왜냐하면 기존 서비스 구현은 Repository 를 직접 이용하여 repo와 session 을 둘다 파라메터로 받기 때문입니다.

# 현재 서비스 함수 구현

def add(
    ref: str,
    sku: str,
    qty: int,
    eta: Optional[date],
    repo: AbstractRepository,
    session: AbstractSession,
) -> None:
    """배치를 추가합니다."""
    repo.add(models.Batch(ref, sku, qty, eta))
    session.commit()

4장에서 FakeSession 과 유사하지만 FakeSession은 sqlalchemy.orm.Session 이라는 외부 라이브러리 구현을 흉내(Mock)냈음에 비해, FakeUnitOfWork 는 우리가 구현한 UnitOfWork 객체를 대상으로 한다는 점에서 큰 발전입니다.

참고: Don’t Mock What You Don’t Own¶

“Don’t mock what you don’t own” is a rule of thumb that forces us to build simple abstractions over messy subsystems. This has the same performance benefit as mocking the SQLAlchemy session but encourages us to think carefully about our designs.

For example, if you mock Session with FakeSession, by coupling to the Session interface, you’re choosing to couple to all the complexity of SQLAlchemy. Instead, we want to choose a simpler abstraction and use that to clearly separate responsibilities.

Our UoW is much simpler than a session, and we feel comfortable with the service layer being able to start and stop units of work.

Using the UoW in the Service Layer¶

앞서 만든 테스트가 성공하도록 UoW를 이용하여 서비스 레이어 함수를 재작성 해봅시다.

[25]:

from typing import Optional
from datetime import date

from app.domain import models

def add(ref: str, sku: str, qty: int, eta: Optional[date], uow: AbstractUnitOfWork) -> None:
    with uow:
        uow.batches.add(models.Batch(ref, sku, qty, eta))
        uow.commit()

@mytest.fixture
def services():
    services_uow = MagicMock()
    services_uow.batch.add = add
    return services_uow

[26]:

mytest.run(test_add_batch)
mytest.run(test_allocate_returns_allocation)

✅ test_add_batch
✅ test_allocate_returns_allocation

테스트가 잘 성공합니다. 서비스 함수들이 이제 UoW만 의존하도록 개선되었습니다.😊

Explicit Tests for Commit/Rollback Behavior¶

commit 과 rollback 이 올바르게 동작하는지도 테스트해봅시다.

[tests/integration/test_uow.py]

[27]:

import pytest

metadata = start_mappers()
mem_engine = init_engine(metadata, 'sqlite://') # using memory db

@mytest.fixture
def get_session():
    return sessionmaker(mem_engine)


@mytest.test
def test_rolls_back_uncommitted_work_by_default(get_session):
    uow = SqlAlchemyUnitOfWork(get_session)
    with uow:
        insert_batch(uow.session, 'batch1', 'MEDIUM-PLINTH', 100, None)

    # Commit 을 안한 경우 실제 DB에 데이터가 반영되지 않습니다.
    new_session = get_session()
    rows = list(new_session.execute("SELECT * FROM batch"))
    assert [] == rows, f'{rows}'

@mytest.test
def test_rolls_back_committed(get_session):
    uow = SqlAlchemyUnitOfWork(get_session)
    with uow:
        insert_batch(uow.session, 'batch1', 'MEDIUM-PLINTH', 100, None)
        uow.session.commit()

    # commit 을 하면 DB 상태가 변경되어야 합니다.
    new_session = get_session()
    [[ref]] = list(new_session.execute('SELECT reference FROM batch'))
    assert 'batch1' == ref, f'{ref}'
    uow.session.execute('DELETE FROM batch')
    uow.session.commit()

@mytest.test
def test_rolls_back_on_error(get_session):
    class MyException(Exception):
        pass

    uow = unit_of_work.SqlAlchemyUnitOfWork(get_session)
    with pytest.raises(MyException):
        with uow:
            insert_batch(uow.session, 'batch1', 'LARGE-FORK', 100, None)
            raise MyException()
            uow.commit()

    new_session = get_session()
    rows = list(new_session.execute('SELECT * FROM "batch"'))
    assert rows == []

✅ test_rolls_back_uncommitted_work_by_default
✅ test_rolls_back_committed
✅ test_rolls_back_on_error

Explicit Versus Implicit Commits¶

UoW 패턴을 구현하는 방법은 여러가지가 있습니다. 이전 구현에서는 명시적으로 uow.commit() 을 실행할 경우만 커밋이 되는 구현이었습니다.

이번에는 UoW 블록을 마칠 때 자동으로 커밋하고, 예외가 발생할 경우만 롤백하는 UoW를 만들어보겠습니다.

[28]:

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

바뀐 UoW 를 이용하면 서비스 함수에서 commit() 을 빼서 좀더 간결히 만들 수 있습니다.

[29]:

def add(ref: str, sku: str, qty: int, eta: Optional[date], uow: AbstractUnitOfWork) -> None:
    with uow:
        uow.batches.add(models.Batch(ref, sku, qty, eta))
        # uow.commit()

하지만 책에서는 명시적으로 commit() 을 호출하는 방법을 권장합니다. 이유는 다음과 같습니다.

코드가 한 줄이 늘더라도 기본적으로 소프트웨어 동작이 안전하도록 보장합니다.
- 소프트웨어의 기본 동작은 “아무것도 변경하지 않는 것” 이기 때문입니다.
이를 통해 소프트웨어의 변경이 일어나는 시점을 쉽게 예측할 수 있도록 합니다.

한편, 롤백은 자동으로 실행하는것을 권장합니다. 명시적인 commit 이 없는한, UoW 가 끝날 때 이전 상태로 돌립니다.. 이 편이 이해하기 쉽고 자연스러운 방식입니다.

Examples: Using UoW to Group Multiple Operations into an Atomic Unit¶

UoW 패턴을 이용한 예제를 몇가지 살펴봅시다.

Example 1: Reallocate¶

재할당 서비스 함수의 경우, 작업중 예외가 발생하면 UoW의 동작 방식에 의해 이전 상태로 자동 롤백됩니다.

모든 유효성 검사와 세부 작업이 다 성공할 경우에만 명시적으로 호출된 commit 함수에 의해 저장소 내용이 변경됩니다.

[30]:

from app.domain.models import OrderLine

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)
        batch.allocate(line)
        uow.commit()

Example 2: Change Batch Quantity¶

배송중 사고로 인해 배치의 수량이 바뀌어야 할 경우 배치에 할당된 주문선을 해제해야 합니다. 이 때 역시 UoW 의 커밋 방식을 이용해 작업중 발생한 예외로 인한 변경이 자동 롤백되도록 합니다.

[31]:

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

Tidying Up the Integration Tests¶

이제 서비스 계층에서 직접적으로 Repository 나 ORM을 이용하지 않기 때문에 test_orm.py 와 test_repository.py 는 구현에 가까워진 Low Gear 테스트가 되었습니다.

하지만 5장에서 살펴보았듯이, Low Gear 테스트는 세부 구현을 이해하는 데 도움이 되기도 합니다. ORM 이나 Repository 에 완전히 새로운 기능이나 복잡한 검증 로직이 들어갈 경우 이들 테스트는 여전히 유용할 수도 있습니다.

테스트 피라미드를 유지하고, 변경에 대해 테스트가 자꾸 깨지는 문제가 발생하지 않도록 불필요한 테스트를 정리하거나 High Gear 테스트로 전환하는 테스트 리팩토링이 필요합니다.

Wrap-Up¶

이번 장에서 우리는 UoW 패턴의 유용성을 보이고자 했습니다. UoW가 추상화 하는 레포지터리간의 원자적 트랜잭선 작업을 파이썬의 멋진 context manager 문법으로 직관적으로 나타낼 수 있었습니다.

UoW 패턴 사용의 장/단점입니다.

Pros

Cons

Nice abstraction over the concept of atomic operations, and the context manager makes it easy to see, visually, what blocks of code are grouped together atomically.

Explicit control over when a transaction starts and finishes, and our application fails in a way that is safe by default.

Nice place to put all your repositories so client code can access them.

Atomicity isn’t only about transactions; it can help us work with events and the message bus.

Most ORMs already have good abstractions around atomicity. SQLAlchemy even has context managers. You can go a long way just passing a session around.

Be careful about things like rollbacks, multithreading, and nested transactions. Perhaps just sticking to what Django or Flask-SQLAlchemy gives you will keep your life simpler.

RECAP¶

The Unit of Work pattern is an abstraction around data integrity
It works closely with the Repository and Service Layer patterns
This is a lovely case for a context manager
SQLAlchemy already implements this pattern