C.W.K.
Stream
Lesson 03 of 04 · published

자기 CRUD 만들기 — 모든 거 cement 하는 end-to-end

~12 min · epilogue, crud, synthesis, capstone

Level 0HTTP Newbie
0 XP0/46 lessons0/12 achievements
0/120 XP to next level120 XP to go0% complete
"REST 에 대해 읽기가 만들기와 같지 않아. Capstone 연습: 이 퀘스트의 모든 트랙 운동하는 완전 CRUD service 만들기. 한 번 해보면 패턴이 반사 됨."

Spec — Resource 하나, endpoint 여덟, 모든 개념

모든 거 운동하는 작은 /items service 만들기:

  • GET /items — cursor pagination + sorting + sparse fieldset + filter 가진 list.
  • POST /items — server 할당 ID 로 생성, 201 + Location 돌려줌.
  • GET /items/{id} — ETag + Cache-Control + 304 conditional GET 가진 read.
  • PUT /items/{id} — optimistic locking 위한 If-Match 가진 full 교체 (충돌에 412).
  • PATCH /items/{id} — 부분 update.
  • DELETE /items/{id} — 제거, 204 돌려줌.
  • POST /items/{id}/publish — action endpoint ('publish' 에 자연 명사 없음).
  • POST /items/jobs — async bulk operation, job-status resource 에 202 + Location 돌려줌.

추가 cross-cutting: CORS, request_id middleware, 구조화된 error envelope, /docs 의 OpenAPI, deprecated endpoint 의 Sunset header, Bearer token auth, rate limit.

빌드 순서

  1. FastAPI app + Pydantic Item model + in-memory dict store. GET /items 가 list 돌려줌.
  2. 201 + Location 과 server-생성 itm_-prefixed ID 가진 POST /items 추가.
  3. GET /items/{id} 추가. 그 다음 ETag 계산 (정렬된 JSON 의 sha256) 과 If-None-Match 일치에 304 추가.
  4. If-Match 가진 PUT /items/{id} (stale ETag 면 412), 부분 위한 PATCH, 204 가진 DELETE 추가.
  5. Cursor pagination (limit + cursor param, has_more 감지에 limit+1 fetch), filter (whitelist), sort (whitelist), fields (whitelist) 추가.
  6. Async 면 202 가진 action endpoint POST /items/{id}/publish 추가.
  7. Async job 패턴 추가: POST /items/jobs 가 202 + Location, GET /items/jobs/{job_id} 가 status 보고.
  8. CORS middleware, correlation-id middleware, custom error envelope 추가.
  9. WWW-Authenticate 가진 401 가지고 FastAPI HTTPBearer dependency 통해 Bearer auth 추가.
  10. slowapi rate limit (IP 당 10/분) 추가. Deprecated /v1/items route 에 Sunset header 추가.

끝나면 Python 200-300 줄로 교과서-깔끔한 REST service ship. 이 퀘스트 모든 트랙 운동.

품질 bar

Endpoint 동작; 이제 polish 체크:

  • curl -v 로 각 endpoint 치기; status code 가 spec 매칭하는지 검증.
  • Idempotency-Key 가진 같은 POST 두 번 보내; item 하나만 생성되는지 검증.
  • Stale ETag vs current 가진 conditional GET 발신; 200 vs 304 검증.
  • Stale If-Match 가진 PUT; 412 검증.
  • Malformed body 보내; error envelope 에 code, message, request_id 있는지 검증.
  • /docs 방문; Swagger UI 가 type 가진 모든 endpoint 보여주는지 검증.
  • Rate limit 치기; Retry-After header 있는지 검증.
  • Deprecated route 치기; Sunset header 검증.

모든 "검증" 이 "yes, protocol 이 내가 생각하는 거 함" 의 순간. 그게 빌드하는 muscle memory.

Capstone 이 project 아냐; 반사 builder. 한 service 에 Location, ETag, If-Match, 412, correlation ID, error envelope, OpenAPI, Sunset header 연결 한 번 하면 다 명백 됨. 다음에 REST API 만들 때 tutorial 안 잡음 — byte 의도적 shape.

대담한 사람 위한 확장

기본 spec 너머 optional 야망:

  • Opaque token 대신 JWT 추가. PyJWT 로 검증; algorithm 핀; claim (sub, exp) 포함.
  • SSE endpoint 추가 일어나면 create stream.
  • WebSocket endpoint 추가 양방향 collaborative 편집.
  • In-memory 대신 Idempotency-Key 위한 dedupe store 로 Redis 추가.
  • OpenTelemetry traceparent 전파 추가 분산 tracing.
  • OpenAPI spec 에서 TypeScript SDK 생성; 작은 React frontend 에서 사용.

각 확장이 트랙 더 깊이 운동. 어느 것도 capstone 에 필수 아님; 다 잘 밟힌 영토.

Cheat sheet 로 cwkPippa

어느 부분 만들다 막히면 cwkPippa source 가 worked 예. FastAPI middleware 순서 막힘? backend/main.py 봐. Streaming response 막힘? backend/adapters/claude.py 봐. Healing / idempotency 막힘? backend/store/conversations.py 봐. Specific 질문 있을 때 동작 코드 읽기가 tutorial 읽기 이김. cwkPippa 가 너 거 베끼라고.

Code

Capstone scaffold — auth, 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 빌드 위해 연습 봐)
Conditional GET (304) + optimistic update (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 에 capstone /items service 빌드. Curl -v 로 모든 endpoint 치기; 각 status code, 각 header (ETag, Location, Cache-Control, Vary, Retry-After, Sunset, X-Request-ID, WWW-Authenticate) 검증. 보너스: 각 endpoint 운동하고 response shape 에 assert 하는 작은 test suite 써 — 미래 변경 위한 regression net 됨. 트리플 보너스: OpenAPI spec 에서 TypeScript SDK 생성하고 그거 end-to-end 쓰는 작은 client script 써.
Hint
코드 블록 의 FastAPI scaffold 로 시작; 점진 확장 (한 번에 모든 거 쓰려 마). 다음 추가 전 curl 로 각 조각 검증. Test suite 이 spec 이 그냥 문서 아닌 — 언제든 재실행 가능한 assertion 이었다 가르쳐. TypeScript SDK 데모가 loop 닫음: OpenAPI spec 하나 → typed client → 동작 앱.

Progress

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

댓글 0

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

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