좋습니다. 이제 문서 → 실제 동작하는 시스템으로 넘어갑니다.
아래는 복붙해서 바로 실행 가능한 AOP 최소 레퍼런스 (FastAPI + Event Store + Replay)입니다.
🚀 AOP Minimal Reference Implementation
목표
event 기록
projection 생성
실패 발생
replay로 복구
1. 아키텍처 (최소 구성)
FastAPI (API)
↓
Command Handler
↓
Event Store (SQLite)
↓
Projection (in-memory)
↓
Replay Engine
2. 프로젝트 구조
Bash
aop_minimal/
├── main.py
├── event_store.py
├── projection.py
├── handlers.py
├── replay.py
└── models.py
3. 코드
3.1 models.py
Python
실행됨
from pydantic import BaseModel
from typing import Optional
import uuid
class Command(BaseModel):
type: str
payload: dict
trace_id: str = str(uuid.uuid4())
class Event(BaseModel):
id: str
type: str
payload: dict
trace_id: str
seq: int
3.2 event_store.py
Python
실행됨
import sqlite3
from models import Event
conn = sqlite3.connect(“events.db”, check_same_thread=False)
cursor = conn.cursor()
cursor.execute(“””
CREATE TABLE IF NOT EXISTS events (
id TEXT,
type TEXT,
payload TEXT,
trace_id TEXT,
seq INTEGER
)
“””)
conn.commit()
def append_event(event: Event):
cursor.execute(
“INSERT INTO events VALUES (?, ?, ?, ?, ?)”,
(event.id, event.type, str(event.payload), event.trace_id, event.seq)
)
conn.commit()
def get_events(trace_id: str):
cursor.execute(
“SELECT id, type, payload, trace_id, seq FROM events WHERE trace_id=? ORDER BY seq”,
(trace_id,)
)
rows = cursor.fetchall()
return rows
3.3 projection.py
Python
실행됨
projections = {}
def apply_event(event):
trace_id = event.trace_id
if trace_id not in projections:
projections[trace_id] = {“status”: “INIT”}
if event.type == “OrderCreated”:
projections[trace_id][“status”] = “CREATED”
elif event.type == “PaymentRequested”:
projections[trace_id][“status”] = “PAYMENT_PENDING”
elif event.type == “PaymentCompleted”:
projections[trace_id][“status”] = “COMPLETED”
elif event.type == “PaymentFailed”:
projections[trace_id][“status”] = “FAILED”
def get_projection(trace_id):
return projections.get(trace_id, {})
3.4 handlers.py
Python
실행됨
import uuid
from models import Event
from event_store import append_event
from projection import apply_event
seq_counter = {}
def next_seq(trace_id):
seq_counter.setdefault(trace_id, 0)
seq_counter[trace_id] += 1
return seq_counter[trace_id]
def emit(event_type, payload, trace_id):
event = Event(
id=str(uuid.uuid4()),
type=event_type,
payload=payload,
trace_id=trace_id,
seq=next_seq(trace_id)
)
append_event(event)
apply_event(event)
return event
def handle_create_order(cmd):
trace_id = cmd.trace_id
emit(“OrderCreated”, cmd.payload, trace_id)
emit(“PaymentRequested”, cmd.payload, trace_id)
# intentionally fail once
if cmd.payload.get(“fail”):
emit(“PaymentFailed”, {}, trace_id)
raise Exception(“Payment failed”)
emit(“PaymentCompleted”, {}, trace_id)
3.5 replay.py
Python
실행됨
from event_store import get_events
from projection import apply_event, projections
def replay(trace_id):
projections[trace_id] = {“status”: “INIT”}
events = get_events(trace_id)
for row in events:
_, type_, payload, trace_id, seq = row
event = type(“Event”, (), {})()
event.type = type_
event.payload = eval(payload)
event.trace_id = trace_id
event.seq = seq
apply_event(event)
return projections[trace_id]
3.6 main.py
Python
실행됨
from fastapi import FastAPI
from models import Command
from handlers import handle_create_order
from projection import get_projection
from replay import replay
app = FastAPI()
@app.post(“/run”)
def run(cmd: Command):
try:
handle_create_order(cmd)
except Exception as e:
return {“status”: “failed”, “trace_id”: cmd.trace_id}
return {“status”: “ok”, “trace_id”: cmd.trace_id}
@app.get(“/projection/{trace_id}”)
def projection(trace_id: str):
return get_projection(trace_id)
@app.post(“/replay/{trace_id}”)
def replay_endpoint(trace_id: str):
result = replay(trace_id)
return {“replayed”: True, “projection”: result}
4. 실행 방법
Bash
pip install fastapi uvicorn pydantic
uvicorn main:app –reload
5. 테스트 시나리오
5.1 정상 실행
Bash
curl -X POST localhost:8000/run \
-H “Content-Type: application/json” \
-d ‘{“type”:”CreateOrder”,”payload”:{}}’
→ projection 확인
Bash
curl localhost:8000/projection/{trace_id}
5.2 실패 유도
Bash
curl -X POST localhost:8000/run \
-H “Content-Type: application/json” \
-d ‘{“type”:”CreateOrder”,”payload”:{“fail”:true}}’
→ FAILED 상태 확인
5.3 Replay 실행
Bash
curl -X POST localhost:8000/replay/{trace_id}
→ 상태 복구 확인
6. 이 코드의 의미 (중요)
이건 production 코드가 아니라:
👉 “AOP의 본질을 보여주는 최소 단위”
여기서 이미 핵심 4개가 들어있습니다:
event 기록
projection 생성
failure 발생
replay로 복구
7. 다음 확장 방향
이걸 기반으로 다음을 붙이면 실제 시스템이 됩니다:
7.1 반드시 추가해야 하는 것
DLQ
retry / backoff
idempotency key
side-effect isolation
event schema versioning
trace / correlation id propagation
7.2 Control Plane
replay request API
preflight simulation
approval workflow
audit log
operator console
7.3 Multi-tenant
tenant_id 추가
quota enforcement
rate limiting
8. 진짜 다음 단계
이제 남은 건 딱 두 개입니다:
1️⃣ Production-grade 구조로 확장
Kafka / Redis Streams
Postgres projection
async worker
distributed trace
2️⃣ 실제 도메인 적용
결제
정산
크롤링
LLM agent orchestration
원하면 다음으로:
👉 “이 코드를 production-grade AOP 구조로 확장하는 설계 + 코드”
👉 “결제/정산 도메인 실제 적용 버전”
까지 바로 이어서 만들어드리겠습니다.