Architecture Patterns with Python(4장)

플라스크 API와 서비스 계층

 

앞서 만든 도메인 서비스와 저장소 인터페이스를 엮어 깔끔한 아키텍처로 리팩토링 해본다.

다음과 같은 계획에 따른다.

 

1. 플라스크를 사용해 allocate 도메인 서비스(1장) 앞에 API 엔드 포인트를 위치한다. 데이터 베이스 세션과 저장소를 연결한다. 그 후 End to End 테스트와 SQL 을 활용한 테스트를 한다.
2. 서비스 계층을 리팩토링해서 플라스크와 도메인 모델 사이에 유스 케이스를 담는 추상화 역할을 할 수 있게 한다.
3. 서비스 계층의 기능을 여러 유형의 파라미터로 실험한다. 

 

데이트베이스 한쪽 끝부터 API 사용하는 반대편 끝까지에 대한 테스트를 엔드투엔드 테스트 라고 한다.

import pytest
import uuid
import requests

import config


def random_suffix():
    return uuid.uuid4().hex[:6]


def random_sku(name=""):
    return f"sku-{name}-{random_suffix()}"


def random_batchref(name=""):
    return f"batch-{name}-{random_suffix()}"


def random_orderid(name=""):
    return f"order-{name}-{random_suffix()}"


@pytest.mark.usefixtures("restart_api")
def test_happy_path_returns_201_and_allocated_batch(add_stock):
    sku, othersku = random_sku(), random_sku("other")
    earlybatch = random_batchref(1)
    laterbatch = random_batchref(2)
    otherbatch = random_batchref(3)
    add_stock(
        [
            (laterbatch, sku, 100, "2011-01-02"),
            (earlybatch, sku, 100, "2011-01-01"),
            (otherbatch, othersku, 100, None),
        ]
    )
    data = {"orderid": random_orderid(), "sku": sku, "qty": 3}
    url = config.get_api_url()

    r = requests.post(f"{url}/allocate", json=data)

    assert r.status_code == 201
    assert r.json()["batchref"] == earlybatch

 

랜덤으로 sku, batch 를 생성하고, 이를 db에 추가하는 부분을 테스트한 코드이다.

이를 만족시키는 API를 아래와 같이 구현할 수 있다.

from flask import Flask, request, jsonify
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

import config
from chapter_01.model import model_final as model
from chapter_02 import orm, repository


orm.start_mappers()
get_session = sessionmaker(bind=create_engine(config.get_postgres_uri()))
app = Flask(__name__)


@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    session = get_session()
    batches = repository.SqlAlchemyRepository(session).list()
    line = model.OrderLine(
        request.json["orderid"],
        request.json["sku"],
        request.json["qty"],
    )

    batchref = model.allocate(line, batches)

    return jsonify({"batchref": batchref}), 201

 

POST method 로 body 를 통해 넘겨 받은 각각의 데이터를 통해서 OrderLine  객체를 생성하고, 이를 배치에 할당하는 allocate 함수를 쓴다. 그 결과로 batch의 reference id 를 반환받아 json 형태로 response 해주는 api 다. 

__eq__ method 에서 ref id 로 비교하도록 구현했기 때문에 테스트 코드에서 r.json()["batchref"] == earlybatch 로 assertion이 가능하다.

 

참고) Batch Class

class Batch:
    def __init__(self, ref: str, sku: str, qty: int, eta: Optional[date]):
        self.reference = ref
        self.sku = sku
        self.eta = eta
        self._purchased_quantity = qty
        self._allocations = set()  # type: Set[OrderLine]

    def __eq__(self, other):
        if not isinstance(other, Batch):
            return False
        return other.reference == self.reference

	...

 

정의된 allocate_endpoint() 는 데이터베이스에 커밋하는 코드가 없다. 이 부분에 대한 테스트가 필요하다. 데이터베이스의 상태를 검사하고, 배치를 모두 소진한 경우 라인 할당이 되지 않아야 하는 사실을 검사하는 테스트다.

@pytest.mark.usefixtures("restart_api")
def test_allocation_are_persisted(add_stock):
    sku = random_sku()
    batch1, batch2 = random_batchref(1), random_batchref(2)
    order1, order2 = random_orderid(1), random_orderid(2)
    add_stock([
        (batch1, sku, 10, "2021-01-01"),
        (batch2, sku, 10, "2021-01-02"),
    ])
    line1 = {"orderid": order1, "sku": sku, "qty": 10}
    line2 = {"orderid": order2, "sku": sku, "qty": 10}
    url = config.get_api_url()
    
    # 첫 번째 주문은 배치 1에 있는 모든 재고를 소진한다.
    r = requests.post(f"{url}/allocate", json=line1)
    assert r.status_code == 201
    assert r.json()["batchref"] == batch1
    
    # 두 번째 주문은 배치 2로 가야 한다.
    r = requests.post(f"{url}/allocate", json=line2)
    assert r.status_code == 201
    assert r.json()["batchref"] == batch2

 

위와 같은 테스트는 데이터베이스 커밋이 되어야 한다. 하지만 테스트에서 실제 데이터 베이스에 커밋이 되길 원하는 경우는 많지 않을 것이다.  또한 데이터베이스 계층에 구현해야 하는 무결성 검사를 추가하고 싶을 때는 어떻게 할 것인가
예를 들면 도메인이 재고가 소진된 SKU 에 대해 예외가 발생하거나, 존재하지 않는 SKU에 대해 예외가 발생하는 경우를 들 수 있다.

우선 예외 케이스에 대한 처리를 플라스크 앱에 추가해야 한다. 이런 식으로 예외가 늘어날 수록 앱은 복잡해지고 E2E 테스트 개수도 늘어난다. 결국엔 역 피라미드형 테스트가 된다.

역 피라미드 테스트(아이스크림 콘)는 코드의 핵심 로직과 관련된 unit test 나 integration 테스트보다 수동적으로 수행하는 테스트가 더 많아지는 anti-parttern 이다. 급하게 코드를 짜거나 여러 이유로 아키텍처에 대한 고민 없이 보여지는 부분에 대해서만 그때 그때 테스트를 수행하게 되는 것이다.

 

현재 구현된 플라스크 앱이 하는 일은 오케스트레이션 이라고 부르는 요소가 상당 부분을 차지한다. 오케스트레이션은 저장소에서 데이터를 가져오고, 데이터베이스 상태에 따라 입력을 검증하며 오류를 처리하고, 성공적인 경우 데이터베이스에 커밋하는 작업을 포함한다.

이런 작업은 API 엔드포인트와는 관련이 없다. 또한 E2E 테스트 대상도 아니다.

 

오케스트레이션 계층이나 유스 케이스 계층이라고 부르는 서비스 계층으로 분리하는 것이 타당하다.

서비스 계층을 단위 테스트로 테스트할 때, 가짜 저장소를 활용하면 매우 유용하다.

 

class FakeRepository(repository.AbstractRepository):
    def __init__(self, batches):
        self._batches = set(batches)
        
    def add(self, batch):
        self._batches.add(batch)
        
    def get(self, reference):
        return next(b for b in self._batches if b.reference == reference)
    
    def list(self):
        return list(self._batches)

 

저장소 패턴을 통해서 구현된 추상화된 저장소를 상속받아 가짜 저장소를 구현할 수 있다.

이를 이용하여 서비스 계층에서 단위 테스트를 구현한다.

우선 allocate 함수를 서비스 계층으로 분리하여 따로 정의한다.

from chapter_01.model import model_final as model
from chapter_01.model.model_final import OrderLine
from chapter_02.repository import AbstractRepository


class InvalidSku(Exception):
    pass


def is_valid_sku(sku, batches):
    return sku in {b.sku for b in batches}


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

 

전형적으로 서비스 계층 함수들은 다음과 같은 단계를 거친다.

1. 저장소에서 어떤 객체들을 가져온다.
2. 현재를 바탕으로 요청을 검사하거나 검증한다.
3. 도메인 서비스를 호출한다.
4. 모든 단계가 정상적으로 실행됐다면 변경한 상태를 저장, 업데이트 한다.

 

가짜 저장소와 분리된 서비스 계층의 allocate 함수를 사용하여, 할당과 예외처리에 대한 테스트를 구현할 수 있다.

import pytest

from chapter_01.model import model_final as model
from chapter_02 import repository
from chapter_04 import services


class FakeRepository(repository.AbstractRepository):
    def __init__(self, batches):
        self._batches = set(batches)

    def add(self, batch):
        self._batches.add(batch)

    def get(self, reference):
        return next(b for b in self._batches if b.reference == reference)

    def list(self):
        return list(self._batches)


class FakeSession:
    committed = False

    def commit(self):
        self.committed = True


def test_returns_allocation():
    line = model.OrderLine("o1", "COMPLICATED-LAMP", 10)
    batch = model.Batch("b1", "COMPLICATED", 100, eta=None)
    repo = FakeRepository([batch])

    result = services.allocate(line, repo, FakeSession())
    assert result == "b1"


def test_error_for_invalid_sku():
    line = model.OrderLine("o1", "NONEXISTENTSKU", 10)
    batch = model.Batch("b1", "NONEXISTENTSKU", 100, eta=None)
    repo = FakeRepository([batch])

    with pytest.raises(services.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
        services.allocate(line, repo, FakeSession())


def test_commits():
    line = model.OrderLine("o1", "OMINOUS-MIRROR", 10)
    batch = model.Batch("b1", "OMINOUS-MIRROR", 100, eta=None)
    repo = FakeRepository([batch])
    session = FakeSession()

    services.allocate(line, repo, session)
    assert session.committed is True

 

데이터베이스 세션을 대신할 FakeSession 정의하고, 위와 같은 서비스 계층 테스트가 가능하다.

 

위의 services.py 에서 정의된 allocate 함수는 타입 힌트를 사용해 AbstractRepository 에 의존함을 밝혔다. 이는 단위 테스트에서 사용된 FakeRepository 와 실제 서비스에서 사용되는 SqlAlchemyRepository 모두 이 함수에 잘 동작함을 뜻한다. 서비스 계층이라는 고수준 모듈이 저수준 모듈인 SqlAlchemyRepository 에 의존하는 것이 아니라, 추상화된 객체에 의존한다. 동시에 테스트를 위한 FakeRepository 와 SqlAlchemyRepository도 AbstractRepository에 의존함으로써 의존성 역전 원칙이 지켜짐을 알 수 있다.

 

수정된 서비스 계층의 함수를 사용하여 플라스크 앱을 수정하면 아래와 같다.

@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    session = get_session()
    repo = repository.SqlAlchemyRepository(session)
    line = model.OrderLine(
        request.json["orderid"],
        request.json["sku"],
        request.json["qty"],
    )
    try:
        batchref = services.allocate(line, repo, session)
    except (model.OutOfStock, services.InvalidSku) as e:
        return jsonify({"message": str(e)}), 400

    return jsonify({"batchref": batchref}), 201

 

변경 전에는 model.allocate 함수를 플라스크 앱에서 직접 import 하여 수행하였으나, 변경된 코드에서는 이를 서비스에 넘긴다.

플라스크 앱의 책임은 표준적인 웹 기능일 뿐이다. 요청 전 상태를 관리하고 POST 파라미터로부터 정보를 파싱하며 상태 코드를 응답하고 JSON 을 처리한다. 모든 오케스트레이션 로직은 서비스 계층에 들어가고, 도메인 로직은 도메인에 그대로 남는다.

 

이제 E2E 테스트는 정상적인 경로와 비정상적 경로, 단 두가지만을 테스트 한다.

@pytest.mark.usefixtures("restart_api")
def test_happy_path_returns_201_and_allocated_batch(add_stock):
    sku, othersku = random_sku(), random_sku("other")
    earlybatch = random_batchref(1)
    laterbatch = random_batchref(2)
    otherbatch = random_batchref(3)
    add_stock(
        [
            (laterbatch, sku, 100, "2011-01-02"),
            (earlybatch, sku, 100, "2011-01-01"),
            (otherbatch, othersku, 100, None),
        ]
    )
    data = {"orderid": random_orderid(), "sku": sku, "qty": 3}
    url = config.get_api_url()

    r = requests.post(f"{url}/allocate", json=data)

    assert r.status_code == 201
    assert r.json()["batchref"] == earlybatch


@pytest.mark.usefixtures("restart_api")
def test_unhappy_path_returns_400_and_error_message():
    unknown_sku, orderid = random_sku(), random_orderid()
    data = {"orderid": orderid, "sku": unknown_sku, "qty": 20}
    url = config.get_api_url()
    r = requests.post(f"{url}/allocate", json=data)
    assert r.status_code == 400
    assert r.json()["message"] == f"Invalid sku {unknown_sku}"

 

플라스크 앱에서 서비스 계층을 분리함으로써, 크게 웹 기능 테스트와 오케스트레이션 관련 요소를 테스트하는 것으로 나눌 수 있다. 웹 기능 테스트는 E2E 로 구현하고, 오케스트레이션 관련 요소 테스트는 메모리에 있는 서비스 계층을 대상으로 테스트 한다.

 

 

서비스 계층은 외부로부터 오는 요청을 처리해 애플리케이션을 제어한다.

- 데이터베이스에서 데이터를 얻는다.

- 도메인 모델을 업데이트한다.

- 변경된 내용을 반영한다.

 

유사한 이름인 도메인 서비스는 도메인 모델에 속하며, 엔티티나 값 객체에 속하지 않는 로직을 부르는 이름이다.

 

 

서비스 계층을 추가하면 다음과 같은 장점이 있다.

• 플라스크 API 엔드포인트가 복잡하지 않고 작성하기 쉬워진다. JSON 파싱이나 정상 경로, 비정상 경로에 따른 올바른 HTTP 코드 반환 등의 웹 기능만 수행한다.
• 역 피라미드를 피할 수 있다. E2E 테스트는 최소화되고, 단위 테스트 위주의 테스트가 가능하다.
• 저장소 패턴 및 가짜 저장소와 조합하면 도메인 계층보다 더 높은 수준에서 테스트를 작성할 수 있다.
• 도메인 로직이 API 와 직접적으로 맞닿아 있지 않으므로, 자유로운 리팩토링이 가능하다.

반면 단점은 아래와 같다.

• 서비스 계층도 결국 또 다른 추상화 계층에 불과하며, 너무 많은 기능을 넣으면 빈약한 도메인(anemic domain) 안티패턴이 생길 수 있다.
  • 빈약한 도메인 : 도메인 객체들에 비즈니스 로직(확인, 계산, 비즈니스 규칙 등)이 거의 없거나 아예 없는 상태를 말한다. 도메인 모델은 핵심 비즈니스 로직을 담고 있어야 하지만 이마저도 서비스 계층으로 옮겨가 정작 도메인 모델의 객체는 행위가 거의 없게 된다.
• 풍부한 도메인 모델로 얻을 수 있는 이익 대부분은 단순히 컨트롤러에서 로직을 뽑아내 모델 계층으로 보내는 것만으로 충분하며, 굳이 계층을 추가할 필요가 없다.

 

이미지 출처 :

https://sqa.stackexchange.com/questions/37623/is-an-inverted-test-pyramid-really-an-anti-pattern

Is an inverted test pyramid really an anti-pattern?