1.
위의 아키텍처를 바탕으로 아래 계획대로 수행
a.
플라스크
b.
allocate라는 도메인 서비스 앞에 API 엔드 포인트를 위치한다.
i.
데이터 베이스 세션과 저장소를 연결한다.
ii.
그 후 End to End 테스트와 SQL 을 활용한 테스트를 한다.
c.
서비스 계층을 리팩토링해서 플라스크와 도메인 모델 사이에 유스 케이스를 담는 추상화 역할을 할 수 있게 한다.
d.
서비스 계층의 기능을 여러 유형의 파라미터로 실험한다.
•
엔드 투 엔드 테스트란? : HTTP를 사용하여 실제 API 엔드포인트와 실제 데이터 베이스를 사용해서 테스트하는 방법
•
테스트를 완벽하게 성공하려면 '빠른 테스트'와 '느린 테스트'로 나눠야 함.
서비스 계층 소개
•
플라스크 앱이 하는 일을 살펴보면, 오케스트레이션이라고 부르는 요소가 상당 부분을 차지한다.
•
오케스트레이션 작업
◦
저장소에서 여러 가지를 가져오고, 데이터베이스 상태에 따라서 입력을 검증하여 오류를 처리하고, 성공적인 경우에는 데이터베이스에 커밋하는 작업을 포함
◦
하지만 위의 작업은 엔드포인트와는 관련이 없고 E2E 테스트의 대상도 아니다
•
코드 예시
import model
from model import OrderLine
from 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
Python
복사
•
전형적으로 서비스 계층 함수들은 다음과 같은 단계를 거친다.
1.
저장소에서 어떤 객체들을 가져온다.
2.
현재를 바탕으로 요청을 검사하거나 검증한다.
3.
도메인 서비스를 호출한다.
4.
모든 단계가 정상적으로 실행됐다면 변경한 상태를 저장, 업데이트 한다.
다시 말하면, 서비스 계층은 외부로부터 오는 요청을 처리해 애플리케이션을 제어
•
데이터베이스에서 데이터를 얻는다.
•
도메인 모델을 업데이트한다.
•
변경된 내용을 반영한다.
아래는 엔드포인트
@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
Python
복사
•
엔드 포인트에서 좀 더 간단해진 것을 확인할 수 있다.
◦
API에서 오케스트레이션 로직을 신경 쓰지 않아도 되므로 코드가 줄어들고 작성하기 쉬워진다.
◦
오케스트레이션 부분이 복잡해지더라도 API에선 특정 유즈 케이스에 맞는 서비스 계층을 호출하고 결과에 대한 예외 처리만 하면 된다.
•
분리된 테스트
◦
웹 API는 실제 HTTP 통신을 사용해서 E2E 테스트를 해야하지만, 오케스트레이션 부분만 테스트 할 땐 통신으로 인한 지연 없이 더 빠른 테스트가 가능
◦
웹 기능 테스트는 E2E 로 구현
◦
오케스트레이션 관련 요소 테스트는 메모리에 있는 서비스 계층을 대상으로 테스트
•
모든 오케스트레이션 로직은 유지 케이스 / 서비스 계층에 들어가고, 도메인 로직은 도메인에 그대로 남는다.
도메인 서비스와 서비스 계층의 차이
서비스 계층
•
서비스는 외부 세계에서 오는 요청을 처리해서 연산을 오케스트레이션한다.
◦
즉 서비스 계층은 데이터베이스에서 데이터를 얻거나, 도메인 모델을 업데이트 하거나, 변경된 내용을 영속화 한다.
•
이는 시스템에서 어떤 연산이 일어날 때마다 해야하는 지루한 작업이며 비즈니스 로직과 분리하면 프로그램을 깔끔하게 유지하는데 도움이 된다.
from typing import List, Optional
class OrderService:
def __init__(self, repository):
self.repository = repository
def create_order(self, order_id: str, customer_id: str, items: List[OrderItem]) -> Order:
new_order = Order(order_id=order_id, customer_id=customer_id, order_date=date.today(), items=items)
self.repository.save(new_order)
return new_order
def add_item_to_order(self, order_id: str, item: OrderItem) -> None:
order = self.repository.get_by_id(order_id)
if not order:
raise ValueError(f'Order with id {order_id} not found')
order.add_item(item)
self.repository.save(order)
def update_order_status(self, order_id: str, new_status: str) -> None:
order = self.repository.get_by_id(order_id)
if not order:
raise ValueError(f'Order with id {order_id} not found')
order.update_status(new_status)
self.repository.save(order)
def get_order_by_id(self, order_id: str) -> Optional[Order]:
return self.repository.get_by_id(order_id)
class InMemoryOrderRepository:
def __init__(self):
self.orders = {}
def save(self, order: Order) -> None:
self.orders[order.order_id] = order
def get_by_id(self, order_id: str) -> Optional[Order]:
return self.orders.get(order_id)
Python
복사
도메인 서비스
•
도메인 서비스
◦
도메인 모델에 속하지만 근본적으로 상태가 있는 엔티티나 값 객체에 속하지 않는 로직을 부르는 이름.
◦
비즈니스 로직을 캡슐화하며 응집력 있는 도메인 객체를 통해 특정 문제 영역을 모델링하는 패턴
◦
예를 들어서
▪
쇼핑 카트 애플리케이션을 만든다면 도메인 서비스로 세금 관련 규칙을 구현한다.
▪
세금을 계산하는 작업은 쇼핑 카트를 업데이트하는 작업과는 별개이다.
▪
모델에서 중요한 부분이지만, 세금만을 위한 영속적인 엔티티를 사용하는 것은 바람직하지 않다.
from pydantic import BaseModel, Field, validator
from typing import List, Optional
from datetime import date
class OrderItem(BaseModel):
product_id: str
quantity: int
price: float
@validator('quantity')
def quantity_positive(cls, value):
if value <= 0:
raise ValueError('Quantity must be positive')
return value
@property
def total_price(self) -> float:
return self.quantity * self.price
class Order(BaseModel):
order_id: str
customer_id: str
order_date: date
items: List[OrderItem] = Field(default_factory=list)
status: str = 'pending'
@property
def total_amount(self) -> float:
return sum(item.total_price for item in self.items)
def add_item(self, item: OrderItem) -> None:
self.items.append(item)
def update_status(self, new_status: str) -> None:
allowed_statuses = ['pending', 'shipped', 'delivered', 'cancelled']
if new_status not in allowed_statuses:
raise ValueError(f'Invalid status: {new_status}')
self.status = new_status
Python
복사
단점
•
서비스 계층은 기존의 컨트롤러-도메인 구조가 점점 복잡해질 때에 고민하는 게 좋다
◦
반대로 복잡하지 않다면 서비스 계층을 넣는 것이 불필요하거나 오히려 안티 패턴이 될 수 있다
▪
빈약한 도메인 패턴은 비즈니스 로직이 도메인 모델이 아닌 서비스 계층에 집중되어 생기는 안티패턴
◦
레포지토리와 달리 서비스는 필수가 아니다…!
정리
•
서비스 계층을 추가하면 다음과 같은 장점이 있다.
•
플라크스 API 엔드포인트가 아주 얇아지고 작성하기 쉬워진다.
◦
플라스크 API 엔드 포인트는 오직 JSON 파싱이나 정상 경로나 비정상 경로에 따른 올바른 HTTP 코드 반환등의 ‘웹 기능’만 수행한다.
•
도메인에 대한 명확한 API를 정의하였다.
◦
이런 API는 자신이 API인지 CLI인지 심지어 테스트인지에 관계없이 어댑터가 도메인 모델 클래스를 몰라도 사용할 수 있는 유스케이스나 진입점 집합이다.
•
서비스 계층을 사용하면 테스트를 ‘높은 기어비’로 작성할 수 있고 도메인 모델을 적합한 형태로 마음껏 리팩터링할 수 있다.
◦
서비스 계층을 활용하면 같은 유스케이스를 제공할 수 있는 한 이미 존재하는 수 많은 테스트를 재작성하지 않고도 새로운 설계를 테스트할 수 있다.
◦
테스트 코드를 "높은 기어비"로 작성한다는 것은 테스트 코드가 빠르고 효율적으로 실행되며, 변경이 필요할 때 쉽게 수정하고 유지보수할 수 있다는 것을 의미
•
작성한 테스트의 피라미드 구조도 좋아보인다.
◦
테스트 상당 부분은 빠른 단위 테스트이며 E2E나 통합 테스트는 최소화 된다.
질문
1.
여러분이 생각하시는 빠른 테스트와 느린 테스트
2.
도메인 객체 혹은 레이어는 남더라도 서비스 레이어는 없어도 되는가