[파이썬으로 살펴보는 아키텍처 패턴] 챕터 12 명령-질의 책임 분리(CQRS)

# src/allocate/entrypoints/flask_app.py@app.route("/allocate", methods=["POST"]) def allocate_endpoint(): try: cmd = commands.Allocate( request.json["orderid"], request.json["sku"], request.json["qty"] ) uow = unit_of_work.SqlAlchemyUnitOfWork() results = messagebus.handle(cmd, uow)# 메시지 버스에서 result 받아서 반환함 batchref = results.pop(0) except InvalidSku as e: return {"message": str(e)}, 400 return {"batchref": batchref}, 201

from allocation import views @app.route("/allocate", methods=["POST"]) def allocate_endpoint(): try: cmd = commands.Allocate( request.json["orderid"], request.json["sku"], request.json["qty"] ) uow = unit_of_work.SqlAlchemyUnitOfWork() messagebus.handle(cmd, uow)# 메시지 버스는 반환 값이 없고 그저 커맨드만 보냄except InvalidSku as e: return {"message": str(e)}, 400 return "OK", 202# 성공시 OK 반환@app.route("/allocations/<orderid>", methods=["GET"]) def allocations_view_endpoint(orderid): uow = unit_of_work.SqlAlchemyUnitOfWork() result = views.allocations(orderid, uow)# 읽기 쿼리를 날리고 결과 값 받아서 반환if not result: return "not found", 404 return jsonify(result), 200

# src/allocation/views.pyfrom allocation.service_layer import unit_of_work def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork): with uow:## 하드코딩된 SQL을 사용 한다면? results = list(uow.session/execute( 'SELECT ol.sku, b.reference' ' FROM allocations AS a' ' JOIN batches AS b ON a.batch_id = b.id' ' JOIN order_lines AS ol ON a.orderline_id = ol.id' ' WHERE orderid = :orderid', dict(orderid=orderid) )) return [{'sku': sku, 'batchred': batchref} for sku, batchref in results]

def test_allocations_view(sqlite_session_factory): uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory) messagebus.handle(commands.CreateBatch("sku1batch", "sku1", 50, None), uow) messagebus.handle(commands.CreateBatch("sku2batch", "sku2", 50, today), uow) messagebus.handle(commands.Allocate("order1", "sku1", 20), uow) messagebus.handle(commands.Allocate("order1", "sku2", 20), uow) # 제대로 데이터를 얻는지 보기 위해 여러 배치와 주문 추가 messagebus.handle(commands.CreateBatch("sku1batch-later", "sku1", 50, today), uow) messagebus.handle(commands.Allocate("otherorder", "sku1", 30), uow) messagebus.handle(commands.Allocate("otherorder", "sku2", 10), uow) # order1에 대해 잘 할당 됐는지 테스트assert views.allocations("order1", uow) == {"sku": "sku1", "batchref": "sku1batch"}, {"sku": "sku2", "batchref": "sku2batch"}, ]

# src/allocation/views.pyfrom allocation.service_layer import unit_of_work def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork): with uow: # 저장소는 product 객체를 반환해 주어진 주문에서 SKU에 해당하는 모든 상품 찾아야함 products = uow.products.for_order(orderid=orderid) # 모든 배치를 가져옴 batches = [b for p in products for b in p.batches] return [# 원하는 주문에 대한 배치만 찾기 위해 배치를 다시 걸러냄 {'sku': b.sku, 'batchref': b.reference} for b in batches if orderid in b.orderids ]

# src/allocation/views.pyfrom allocation.service_layer import unit_of_work def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork): with uow:# orm 사용 batches = uow.session.query(model.Batch).join( model.OrderLine, model.Batch._allocations ).filter( model.OrderLine.orderid == orderid ) return [ {'sku': b.sku, 'batchref': b.batchref} for b in batches ]

from allocation.service_layer import unit_of_work def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork): with uow: results = uow.session.execute( """ # 가장 빠른 질의문으로 변경됨 SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid """, dict(orderid=orderid), ) return [dict(r) for r in results]

# src/allocation/adapters/orm.py allocations_view = Table( "allocations_view", metadata, Column("orderid", String(255)), Column("sku", String(255)), Column("batchref", String(255)), )

# src/allocation/service_layer/messagebus.py# Allocated 에 대한 새 핸들러 추가 EVENT_HANDLERS = { events.Allocated: [# 할당 handlers.publish_allocated_event, handlers.add_allocation_to_read_model, ], events.Deallocated: [# 할당 해제 handlers.remove_allocation_from_read_model, handlers.reallocate, ], events.OutOfStock: [handlers.send_out_of_stock_notification], }

# src/allocation/service_layer/handler.py# 할당시 업데이트def add_allocation_to_read_model( event: events.Allocated, uow: unit_of_work.SqlAlchemyUnitOfWork, ): with uow: uow.session.execute( """ INSERT INTO allocations_view (orderid, sku, batchref) VALUES (:orderid, :sku, :batchref) """, dict(orderid=event.orderid, sku=event.sku, batchref=event.batchref), ) uow.commit() # 할당 해제 시 삭제def remove_allocation_from_read_model( event: events.Deallocated, uow: unit_of_work.SqlAlchemyUnitOfWork, ): with uow: uow.session.execute( """ DELETE FROM allocations_view WHERE orderid = :orderid AND sku = :sku """, dict(orderid=event.orderid, sku=event.sku), ) uow.commit()

# src/allocation/service_layer/handler.py# 레디스 읽기 모델을 업데이트하는 핸들러def add_allocation_to_read_model(event: events.Allocated, _): redis_eventpublisher.update_readmodel(event.orderid, event.sku, event.batchref) def remove_allocation_from_read_model(event: events.Deallocated, _): redis_eventpiblisher.update_readmodel(event.orderid, event.skum None)

# src/allocation/adapters/redis_eventpublisher.py# 레디스 모듈의 도우미 함수들은 한 줄 짜리 함수들이다def updated_readmodel(orderid, sku, batchref): r.hset(orderid, sku, batchref) def get_readmodel(orderid): return r.hgetall(orderid)

# src/allocation/views.py# 레디스에 맞춰 변경한 뷰def allocations(orderid): batches = redis_eventpublisher.get_readmodel(orderid) return [ {"batcheref": b.decode(), "sku": s.decode()} for s, b in batched.items() ]