파이썬 아키텍처 적용 스터디: chap.4 서비스 계층

티스토리 뷰

Studies/파이썬 아키텍처

파이썬 아키텍처 적용 스터디: chap.4 서비스 계층

prograsshopper_ 2024. 3. 24. 16:45

Chapter.4 첫번째 유스 케이스: 플라스크 API와 서비스 계층

Goal

오케스트레이션 로직, 비즈니스 로직, 연결 코드간의 차이에 대한 설명
서비스 계층 패턴 소개 - 워크 플로를 조정하고 시스템의 유스 케이스를 정의하는 계층
테스트 논의 - 단순 도메인 모델뿐만 아니라 유스케이스의 전체 워크 플로 테스트
- 유스 케이스: 시스템의 하나 이상의 액터 또는 이해관계자에게 관측 가능한 결과를 산출하는 시스템에 의해 수행되는 일련의 활동의 명세
API를 추가해 도메인 모델에 대한 진입점 역할 부여

사용자 피드백을 위해 MVP를 만드는 상황이라 가정

도메인 중심부

도메인 서비스 주문을 할당

저장소 인터페이스 데이터 영구 저장

깔끔한 아키텍처로 리팩토링하기 위한 계획
- 플라스크로 allocate 도메인 서비스 앞에 API 엔드포인트 위치. 데이터베이스 세션과 저장소 연결. 엔드투엔드 테스트와 빠르게 만든 몇가지 SQL 문을 활용해 테스트
- 서비스 계층을 플라스크와 도메인 모델 사이에 유스케이스를 담는 추상화 역할을 할 수 있게 한다.
- 서비스 계층의 기능을 여러 유형의 파라미터로 실험 - 원시 데이터 타입을 사용해서 서비스 계층의 클라이언트를 모델 계층으로부터 분리할 수 있음을 증명

엔드투엔드(end-to-end, E2E) 테스트

애플리케이션의 흐름을 처음부터 끝까지 테스트하는 것 [출처]
- 실제 API 엔드포인트와 실제 데이터베이스를 사용하는 테스트

@pytest.mark.usefixtures("restart_api")
def test_api_returns_allocations(add_stock):
	#  uuid 모듈을 사용하여 난수 문자열을 만드는 작은 도우미 함수
    sku, othersku = random_sku(), random_sku("other")
    earlybatch = random_batchref(1)
    laterbatch = random_batchref(2)
    otherbatch = random_batchref(3)
    
    # SQL을 사용해 데이터베이스에 행을 삽입하는 과정을 숨겨주는 도우미 픽스처
    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() # config: 설정 정보를 저장하는 모듈

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

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

가장 뻔한 방법으로 구현된 플라스크 앱 (코드 전문)

#  데이터베이스에 커밋하는 코드가 없다

@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"],
    )

    batch_ref = model.allocate(line, batches)

    return {"batchref": batchref}, 201

아키텍처 우주인(architecture astronaut)?

지나치게 소프트웨어 설계를 뒷받침하는 추상적인 아이디어에 중점을 두는 개인 [출처]
더보기
위대한 사상가는 문제에 대해 생각할 때 패턴을 보기 시작합니다. 그들은 일반적인 패턴이 있다는 것을 깨닫습니다. 그것은 이미 한 수준의 추상화입니다. 그런 다음 한 단계 더 올라갑니다.새롭고 더 높고 더 넓은 추상화를 발명했지만 이제는 매우 모호해지고 아무도 더 이상 그들이 무슨 말을 하는지 알지 못합니다. … 언제 멈춰야 할지 모르고, 모두 훌륭하고 훌륭하지만 실제로는 아무 의미도 없는 터무니없고 모든 것을 포괄하는 높은 수준의 그림을 만들어냅니다.

할당이 영속화되었는지 검사하는 테스트

데이터베이스 상태를 나중에 검사하는 식이거나, 이미 할당해서 배치를 모두 소진한 경우 두번째 라인 할당이 되지 않아야한단 사실을 검사하는 테스트

@pytest.mark.usefixtures("restart_api")
def test_allocations_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, '2011-01-01'),
        (batch2, sku, 10, '2011-01-012'),
    ])
    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

권장되지 않는 방식이다. 데이터베이스 계층에 구현해야하는 데이터 무결성 검사는 도메인 서비스 호출 전에 이루어져야한다.

E2E 수준에서 이루어지는 추가 테스트

@pytest.mark.usefixtures("restart_api")
def test_400_message_for_out_of_stock(add_stock):
    # 재고보다 더 많은 단위 할당 시도
    sku, small_batch, large_order = random_sku(), random_batchref(), random_orderid()
    add_stock([
        (small_batch, sku, 10, '2011-01-01')
    ])
    data = {"orderid": large_order, "sku": 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'Out of stock for sku {sku}'


@pytest.mark.usefixtures("restart_api")
def test_400_message_for_invalid_sku():
    # sku가 존재하지 않는다 - out of stock 호출하지않음
    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}"

복잡해지기 시작하는 플라스크 앱

@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    request.json["orderid"], request.json["sku"], request.json["qty"],
    )

    batch_ref = model.allocate(line, batches)
    if not is_valid_sku(line.sku, batches):
        return jsonify({'message': f'Invalid sku {line.sku}'}), 400

    try:
        batchref = model.allocate(line, batches)
    except model.OutOfStock as e:
        return {"message": str(e)}, 400

    return {"batchref": batchref}, 201

E2E테스트 개수가 제어가 가능한 수준을 넘었다 → 역피라미드형 테스트 (아이스크림 콘 모델)

대부분의 테스트 노력을 수동 테스트에 투입하면 발생, 확장에 어려움이 있다. [출처]

서비스계층 소개와 서비스 계층 테스트용 FakeRepository 사용

위의 플라스크 앱은 웹 API 엔드포인트와 상관없는 오케스트레이션이 상당부분 차지 -> 서비스 계층으로 분리 필요

오케스트레이션: 저장소에서 여러가지를 가져오고 데이터베이스 상태에 따라 입력을 검증하며 오류를 처리하고 성공적인 경우 데이터를 데이터베이스에 커밋하는 작업을 포함

서비스 계층: 오케스트레이션 계층이나 유스케이스 계층으로도 부름

가짜 저장소를 활용해 서비스 계층을 빠르게 단위테스트로 테스트가 가능하다 [코드 전문]

# 가짜 저장소를 활용한 단위테스트
def test_returns_allocation():
    line = model.OrderLine("o1", "COMPLICATED-LAMP", 10)
    batch = model.Batch("b1", "COMPLICATED-LAMP", 100, eta=None)
    repo = FakeRepository([batch]) # 테스트에 사용할 Batch 객체 저장

    result = services.allocate(line, repo, FakeSession()) # services.allocate: 서비스 계층 함수
    assert result == "b1"


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

    with pytest.raises(services.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
    	# FakeSession: 데이터베이스 세션을 가짜로 제공
        services.allocate(line, repo, FakeSession())

전형적인 서비스 함수

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

단계

저장소에서 어떤 객체를 가져온다
현재 에플리케이션이 아는 세계를 바탕으로 요정을 검사하거나 검증한다
도메인 서비스 호출
모든 단계가 정상으로 실행이라면 변경 상태를 저장하거나 업데이트

깔끔해진 플라스크 앱

@app.route("/allocate", methods=["POST"])
def allocate_endpoint():
    session = get_session()
    repo = repository.SqlAlchemyRepository(session).list()
    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

효과

플라스크 앱의 책임은 표준적인 웹기능일뿐이다.
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}"

해당 장에서 서비스라 부르는 두 요소

애플리케이션 서비스(서비스 계층)
- 외부세계에서 오는 요청을 처리해 연산을 오케스트레이션한다.
- 다음 단계를 수행해서 애플리케이션을 제어한다.
  1. 데이터베이스 에서 데이터를 얻는다
  2. 도메인 모델을 업데이트한다
  3. 변경된 내용을 영속화한다.
도메인 서비스
- 도메인 모델에 속하지만 근본적으로 상태가 있는 엔티티나 값객체에 속하지 않는 로직을 부르는 이름
- 모델에서 중요한 부분이지만 영속적인 엔티티를 쓸 필요는 없을때

정리된 디렉터리 구조

어댑터: 외부 I/O를 감싸는 추상화를 넣는다.

진입점: 애플리케이션을 제어하기 시작하는 지점

서비스 계층의 장점

플라스크 API의 엔드포인트가 아주 얇아지고 작성하기 쉬워진다
도메인에 대한 명확한 API를 정의했다. 이런 API는 자신이 무엇인지와 관계없이 어댑터가 도메인 모델 클래스를 몰라도 사용가능한 유스케이스나 진입점 집합이다
서비스 계층을 사용하면 테스트를 높은 기어비로 작성할 수 있고 도메인 모델을 적합한 형태로 마음껏 리팩터링할 수 있다.
- 기어비: 서로 맞물리는 두 기어에서 큰 기어의 톱니수에서 작은 기어의 톱니수로 나누는 값
작성한 테스트의 피라미드도 양호하다

저작자표시 비영리 변경금지

'Studies > 파이썬 아키텍처' 카테고리의 다른 글

파이썬 아키텍처 적용 스터디: chap.7 애그리게이트와 일관성 경계 (0)	2024.04.07

공지사항

최근에 올라온 글

최근에 달린 댓글

Total

Today

Yesterday

링크

TAG more

« 2024/05 »
일	월	화	수	목	금	토
			1	2	3	4
5	6	7	8	9	10	11
12	13	14	15	16	17	18
19	20	21	22	23	24	25
26	27	28	29	30	31

글 보관함

코딩베짱이의 무언가 생산적인 일을 하는 블로그

티스토리 뷰