[파이썬으로 살펴보는 아키텍처 패턴] 챕터 4 첫번째 유스 케이스: 플라스크 API와 서비스 계층

@pytest.mark.usefixtures('restart_api') def test_api_returns_allocation(add_stock): sku, othersku = random_sku(), random_sku('other')# 1 earlybatch = random_batchref(1) laterbatch = random_batchref(2) otherbatch = random_batchref(3) add_stock([# 2 (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()# 3 r = requests.post(f'{url}/allocate' , json=data) assert r.status code == 201 assert r.json()['batchref'] == earlybatch

from flask import Flask, request from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import config import model import orm import repository import services 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() repo = repository.SqlAlchemyRepository(session) line = model.OrderLine( request.json["orderid"], request.json["sku"], request.json["qty"], ) batchref = services.allocate(line, repo, session) return {"batchref": batchref}, 201

@pytest.mark.usefixtures("restart_api") def test_allocations_are_persisted(add_stock): sku = random_sku() batch1, batch2 = random_batchref(l), random_batchref(2) order1, order2 = random_orderid(l), random_orderid(2) add_stock( [ (laterbatch, sku, 100, "2011-01-02"), (earlybatch, sku, 100, "2011-01-01"), (otherbatch, othersku, 100, None), ] ) 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

@pytest.mark.usefixtures('restart_api') def test_400_message_for_out_of_stock(add_stock):# 1 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():# 2 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(): session = get_session() repo = repository.SqlAlchemyRepository(session) line = model.OrderLine( request.json["orderid"], request.json["sku"], request.json["qty"], ) if not is_valid_sku(line.sku, batches): return jsonify({’message’: f’Invalid sku {line.sku}’}), 400 try: batchref = services.allocate(line, repo, session) except (model.OutOfStock, services.InvalidSku) as e: return {"message": str(e)}, 400 return {"batchref": batchref}, 201

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-LAMP", 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", "AREALSKU", 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

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()# 1if not is_valid_sku(line.sku, batches):# 2raise InvalidSku(f"Invalid sku {line.sku}") batchref = model.allocate(line, batches)# 3 session.commit()# 4return batchref

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

@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}"

├── config.py ├── domain.py │ ├── __init__.py │ └── model.py ├── sevice_layer.py │ ├── __init__.py │ └── services.py ├── adapter.py │ ├── __init__.py │ ├── repository.py │ └── orm.py ├── entrypoints.py │ ├── __init__.py │ ├── flask_app.py └── tests.py