본문 바로가기
C.W.K.
Stream
Lesson 03 of 04 · published

직접 만드는 CRUD — 모든 개념을 하나의 서비스에

~12 min · epilogue, crud, synthesis, capstone

Level 0HTTP 입문자
0 XP0/46 lessons0/12 achievements
0/120 XP to next level120 XP to go0% complete
"REST를 읽는 것과 직접 구현하는 것은 다르다. 이 종합 과제에서는 퀘스트의 모든 트랙을 활용해 완전한 CRUD 서비스를 만든다. 전체 과정을 한 번 완주하면 각 패턴의 연결 관계가 자연스럽게 떠오르게 된다."

명세 — 리소스 하나, 엔드포인트 아홉 개, 핵심 개념 전체

핵심 개념을 한꺼번에 연습할 수 있는 작은 /items 서비스를 구현한다.

  • GET /items — cursor pagination, 정렬, 일부 필드 선택, 필터를 지원하는 목록 조회.
  • POST /items — 서버가 할당한 ID로 리소스를 생성하고 201과 Location을 반환한다. Idempotency-Key로 중복 생성도 막는다.
  • GET /items/{id} — ETag, Cache-Control, 304 조건부 GET을 지원하는 단건 조회.
  • PUT /items/{id} — 낙관적 동시성 제어를 위해 If-Match를 요구하는 전체 교체이며 충돌 시 412를 반환한다.
  • PATCH /items/{id} — 리소스의 일부 필드를 수정한다.
  • DELETE /items/{id} — 리소스를 제거하고 204를 반환한다.
  • POST /items/{id}/publish — publish를 자연스러운 명사형 리소스로 표현하기 어려울 때 사용하는 동작 엔드포인트다.
  • POST /items/jobs — 비동기 대량 작업을 시작하고 job-status 리소스를 가리키는 202와 Location을 반환한다. GET /items/jobs/{job_id}는 그 작업의 상태를 조회한다.

공통 기능으로 CORS, request ID 미들웨어, 구조화된 오류 응답, /docs의 OpenAPI 문서, 지원 중단 엔드포인트의 Sunset 헤더, Bearer 토큰 인증, rate limit도 추가한다.

구현 순서

  1. FastAPI 앱, Pydantic Item 모델, 메모리 내 dictionary 저장소를 준비한다. GET /items에서 목록을 반환하도록 구현한다.
  2. 서버가 생성한 itm_ 접두사 ID와 Location 헤더를 201 응답에 담는 POST /items를 추가한다. Idempotency-Key를 요구하고, 같은 키와 같은 payload가 다시 오면 새 항목을 만들지 않고 기존 생성 결과를 반환한다.
  3. GET /items/{id}를 추가한다. 이어서 정렬된 JSON의 sha256으로 ETag를 계산하고, If-None-Match가 일치하면 304를 반환한다.
  4. If-Match를 요구하는 PUT /items/{id}를 추가해 오래된 ETag에 412를 반환한다. 부분 수정을 위한 PATCH와 204를 반환하는 DELETE도 추가한다.
  5. limit과 cursor 매개변수를 사용하는 cursor pagination을 구현한다. has_more를 판단하기 위해 limit+1개를 조회하고, filter, sort, fields에는 whitelist를 적용한다.
  6. 202를 반환하는 비동기 동작 엔드포인트 POST /items/{id}/publish를 추가한다.
  7. 비동기 job 패턴을 추가한다. POST /items/jobs는 202와 Location을 반환하고, GET /items/jobs/{job_id}는 처리 상태를 보고한다.
  8. CORS 미들웨어, correlation ID 미들웨어, 사용자 정의 구조화된 오류 응답을 추가한다.
  9. FastAPI HTTPBearer dependency를 사용해 Bearer 인증을 구현하고, 인증 실패 시 WWW-Authenticate 헤더와 함께 401을 반환한다.
  10. slowapi로 IP당 분당 10회로 rate limit을 설정한다. 지원이 중단된 /v1/items route에는 Sunset 헤더를 추가한다.

완성된 결과는 약 200~300줄의 Python 코드로 구성된 단정한 REST 서비스가 된다. 이 퀘스트의 모든 트랙을 한 번씩 직접 적용하는 종합 연습이다.

품질 기준

엔드포인트가 동작하면 다음 기준으로 구현을 점검한다.

  • curl -v로 엔드포인트를 하나씩 호출하고 상태 코드가 명세와 일치하는지 확인한다.
  • 같은 Idempotency-Key로 POST를 두 번 보내 항목이 하나만 생성되는지 확인한다.
  • 오래된 ETag와 최신 ETag로 조건부 GET을 보내 200과 304가 정확히 구분되는지 확인한다.
  • 오래된 If-Match로 PUT을 보내 412가 반환되는지 확인한다.
  • 잘못된 본문을 보내 구조화된 오류 응답에 code, message, request_id가 모두 포함되는지 확인한다.
  • /docs에서 Swagger UI가 모든 엔드포인트와 타입을 올바르게 표시하는지 확인한다.
  • 의도적으로 rate limit을 초과해 Retry-After 헤더가 반환되는지 확인한다.
  • 지원이 중단된 route를 호출해 Sunset 헤더가 반환되는지 확인한다.

각 확인 과정은 프로토콜이 설계한 대로 동작한다는 사실을 직접 검증하는 단계다. 이 경험이 쌓이면 개별 개념이 실무적인 감각으로 연결된다.

이 종합 과제는 단순한 프로젝트가 아니라 설계 반사 신경을 만드는 훈련이다. 하나의 서비스에 Location, ETag, If-Match, 412, correlation ID, 구조화된 오류 응답, OpenAPI, Sunset 헤더를 직접 연결하면 각 개념의 역할이 선명해진다. 이후 REST API를 설계할 때는 모든 바이트와 헤더의 의도를 설명할 수 있게 된다.

추가 도전 과제

기본 명세를 완성한 뒤에는 다음 확장을 선택적으로 시도할 수 있다.

  • 불투명한 토큰 대신 JWT 추가. PyJWT로 검증하되 알고리즘을 명시적으로 고정하고, claims에 sub와 exp를 포함한다.
  • SSE 엔드포인트 추가 항목이 생성될 때 이벤트를 스트리밍한다.
  • WebSocket 엔드포인트 추가 양방향 협업 편집을 구현한다.
  • 메모리 저장소 대신 Redis를 Idempotency-Key 중복 제거 저장소로 사용한다.
  • OpenTelemetry traceparent 전파 추가 분산 추적 정보를 전달한다.
  • OpenAPI 명세에서 TypeScript SDK를 생성해 작은 React 프런트엔드에서 사용한다.

각 확장은 관련 트랙을 한 단계 더 깊이 연습하는 선택 과제다. 종합 과제의 필수 조건은 아니며, 필요한 영역을 골라 확장하면 된다.

동작하는 참고 사례로서의 cwkPippa

구현 중 막히는 부분이 있다면 cwkPippa 소스 코드를 동작하는 사례로 참고할 수 있다. FastAPI 미들웨어 순서는 backend/main.py, 스트리밍 응답은 backend/adapters/claude.py, 복구와 멱등성은 backend/store/conversations.py에서 살펴볼 수 있다. 구체적인 질문이 생겼을 때는 입문서를 처음부터 다시 읽기보다 실제 운영 코드를 확인하는 편이 효과적이다. 코드를 그대로 복사하기보다 패턴이 현실에서 어떻게 연결되는지 분석하는 것이 목적이다.

Code

종합 과제 scaffold — 인증, CORS, ETag 계산, 201과 Location·python
# Capstone scaffold — 뼈에 ~80 줄
import hashlib, json, secrets, time, uuid
from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Response, Header, status, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field

app = FastAPI(title='Items API', version='1.0.0')
app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:5173'],
    allow_credentials=True, allow_methods=['*'], allow_headers=['*'],
)
bearer = HTTPBearer()
TOKENS = {'tok_valid_xyz': 'u_42'}

class Item(BaseModel):
    id: str = Field(..., pattern='^itm_')
    name: str
    description: str | None = None
    version: int

_ITEMS: dict[str, dict] = {}
_JOBS: dict[str, dict] = {}

def compute_etag(d: dict) -> str:
    body = json.dumps(d, sort_keys=True, separators=(',', ':')).encode()
    return f'"{hashlib.sha256(body).hexdigest()[:12]}"'

async def current_user(creds: HTTPAuthorizationCredentials = Depends(bearer)):
    uid = TOKENS.get(creds.credentials)
    if not uid:
        raise HTTPException(401, detail='invalid token',
                            headers={'WWW-Authenticate': 'Bearer'})
    return uid

@app.post('/items', status_code=201)
async def create_item(payload: dict, response: Response, uid=Depends(current_user)):
    item_id = 'itm_' + secrets.token_urlsafe(8)
    item = {'id': item_id, 'version': 1, **payload}
    _ITEMS[item_id] = item
    response.headers['Location'] = f'/items/{item_id}'
    response.headers['ETag']     = compute_etag(item)
    return item

# ... GET / PUT / PATCH / DELETE / cursor pagination / action / async 계속
#     (full 빌드 위해 연습 봐)
조건부 GET(304)과 낙관적 수정(412) 연결·python
# Optimistic update + conditional GET 조합 (더 어려운 절반)
from fastapi import Header

@app.get('/items/{item_id}')
async def read_item(item_id: str, response: Response,
                    if_none_match: str | None = Header(None, alias='If-None-Match'),
                    uid=Depends(current_user)):
    item = _ITEMS.get(item_id)
    if not item:
        raise HTTPException(404)
    etag = compute_etag(item)
    response.headers['ETag'] = etag
    response.headers['Cache-Control'] = 'private, max-age=60, must-revalidate'
    response.headers['Vary'] = 'Authorization'
    if if_none_match == etag:
        response.status_code = 304
        return None
    return item

@app.put('/items/{item_id}')
async def replace_item(item_id: str, payload: dict, response: Response,
                       if_match: str | None = Header(None, alias='If-Match'),
                       uid=Depends(current_user)):
    existing = _ITEMS.get(item_id)
    if not existing:
        raise HTTPException(404)
    current_etag = compute_etag(existing)
    if if_match and if_match != current_etag:
        raise HTTPException(412, detail='ETag 불일치; refetch 후 재시도')
    new = {**existing, **payload, 'version': existing['version'] + 1}
    _ITEMS[item_id] = new
    response.headers['ETag'] = compute_etag(new)
    return new
각 요구사항 검증 — curl 기반 smoke test·bash
# 모든 요구사항에 대해 capstone 검증

# Create 가 201 + Location 돌려줌
curl -i -X POST http://localhost:8000/items -H 'Authorization: Bearer tok_valid_xyz' \
  -H 'Content-Type: application/json' -d '{"name":"first"}'

# Response 에서 ETag 저장
ETAG=$(curl -s -i http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' | grep -i etag | awk '{print $2}' | tr -d '\r')

# Conditional GET — current ETag 면 304
curl -i http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' -H "If-None-Match: $ETAG"

# Optimistic update — stale ETag 면 412
curl -i -X PUT http://localhost:8000/items/itm_xyz -H 'Authorization: Bearer tok_valid_xyz' \
  -H 'Content-Type: application/json' -H 'If-Match: "v99-stale"' -d '{"name":"updated"}'

# OpenAPI / Swagger UI 가용
open http://localhost:8000/docs

External links

Exercise

FastAPI로 종합 과제의 /items 서비스를 구현한다. curl -v로 모든 엔드포인트를 호출하고, 각 상태 코드와 헤더(ETag, Location, Cache-Control, Vary, Retry-After, Sunset, X-Request-ID, WWW-Authenticate)가 올바르게 반환되는지 확인한다. 보너스 과제로 각 엔드포인트를 호출하고 응답 형태를 검사하는 작은 자동화 test suite를 작성한다. 이 suite는 이후 변경으로부터 동작을 보호하는 회귀 test가 된다. 추가 보너스 과제로 OpenAPI 명세에서 TypeScript SDK를 생성하고, 그 SDK를 처음부터 끝까지 사용하는 작은 클라이언트 script를 작성한다.
Hint
코드 블록의 FastAPI scaffold에서 시작해 기능을 하나씩 확장한다. 모든 기능을 한 번에 작성하지 말고, 다음 기능을 추가하기 전에 curl로 현재 구현을 검증한다. 자동화된 test suite는 명세를 언제든 다시 실행할 수 있는 검사로 바꾼다. TypeScript SDK 예제까지 완성하면 OpenAPI 명세에서 타입이 지정된 클라이언트와 동작하는 앱으로 이어지는 전체 흐름을 확인할 수 있다.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고
💛 by 피파warm

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.