8단계 — FastAPI 실무 패턴

1단계의 단일 main.py, 2단계의 폴더 구조 — 이제 그 둘을 실무에서 깨지지 않는 API 로 합쳐요. 라우터 조직 → 비동기 → 검증 → 의존성 → 에러 → 속도 제한 순서로 한 번에.

라우터 조직 — 한 모듈 한 리소스

API 가 자라면 main.py 에 라우트를 직접 적는 방식은 금방 한계가 와요. 리소스별로 파일을 나누고 APIRouter 로 묶어요.

# routers/articles/articles.py
from fastapi import APIRouter

router = APIRouter(prefix="/articles", tags=["articles"])

@router.get("/")
async def list_articles():
    ...

@router.get("/{article_id}")
async def get_article(article_id: int):
    ...

# main.py
from fastapi import FastAPI
from routers.articles import articles
from routers.health import health

app = FastAPI()
app.include_router(articles.router)
app.include_router(health.router)

routers/{도메인}/{리소스}.py — 한 파일에 한 리소스. main.py 는 include_router 만 모아 둬요. 한 도메인의 변경이 다른 파일을 건드리지 않아요.

비동기 — 이벤트 루프를 막지 않기

엔드포인트는 async def 로 적어요. 그래야 ASGI 이벤트 루프 위에서 동시 처리돼요. 문제는 async def 안에서 동기 I/O 를 호출하는 자리예요.

import requests

# ❌ async 안의 블로킹 호출 — 루프 전체가 멈춤
@router.get("/external")
async def call_external():
    return requests.get("https://api.example.com/data").json()

requests.get() 은 응답이 올 때까지 스레드를 잡아 둬요. 이벤트 루프 위에서 이러면 다른 모든 요청 이 함께 멈춰요. 비동기 클라이언트로 바꿔요.

import httpx

# ✅ await 로 양보 — 그 사이 다른 요청 처리
@router.get("/external")
async def call_external():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com/data")
        return r.json()

피할 수 없는 동기 호출 (오래된 라이브러리 등) 은 스레드 풀로 오프로드해요.

import asyncio

@router.get("/legacy")
async def legacy():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_fn)
    return result

검증 — Pydantic schemas.py

요청·응답 모델은 핸들러 안의 인라인 dict 로 두지 말고 schemas.py 한 곳에 모아요. 라우터마다 모델을 다시 정의하면 같은 스키마가 갈라져요.

# schemas.py
from pydantic import BaseModel, Field, field_validator

class ArticleIn(BaseModel):
    title: str = Field(..., max_length=120, description="글 제목")
    slug: str = Field(..., pattern=r"^[a-z0-9-]+$", description="URL 슬러그")
    body: str = Field(..., min_length=1, description="본문")

    @field_validator("title")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("제목은 공백일 수 없습니다")
        return v

class ArticleOut(BaseModel):
    id: int
    title: str
    slug: str

라우터는 이 모델을 임포트해서 타입 힌트로만 쓰면 끝이에요.

# routers/articles/articles.py
from schemas import ArticleIn, ArticleOut

@router.post("/", response_model=ArticleOut, status_code=201)
async def create_article(payload: ArticleIn):
    ...

response_model 을 명시하면 응답 직렬화 스키마가 입력과 분리돼요 — 비밀 필드가 새는 걸 막는 가드 역할도 해요. 검증이 실패하면 FastAPI 가 자동으로 422 와 어느 필드가 왜 틀렸는지 를 응답해요.

의존성 주입 — Depends

DB 연결·인증처럼 여러 엔드포인트가 공유하는 로직은 Depends 로 주입해요. 핸들러마다 psycopg2.connect() 를 직접 부르지 않아요.

# deps.py
from db.connection import get_conn

def get_db():
    with get_conn() as conn:
        yield conn

# routers/articles/articles.py
from fastapi import Depends

@router.get("/{article_id}", response_model=ArticleOut)
async def get_article(article_id: int, db = Depends(get_db)):
    with db.cursor() as cur:
        cur.execute("SELECT id, title, slug FROM articles WHERE id = %s", (article_id,))
        row = cur.fetchone()
    ...

yield 로 setup/teardown 을 한 함수에 묶는 패턴은 컨텍스트 매니저와 같아요. 같은 요청 안에서 같은 의존은 한 번만 평가돼요 (캐시). 테스트할 때는 app.dependency_overrides 로 가짜 의존으로 갈아끼우면 돼요.

에러 — HTTPException + 글로벌 핸들러

실패를 return {"error": "..."} 로 돌려주면 상태 코드는 여전히 200 이고 스키마도 제각각이 돼요. 실패는 예외로 올려요.

from fastapi import HTTPException

@router.get("/{article_id}", response_model=ArticleOut)
async def get_article(article_id: int, db = Depends(get_db)):
    row = fetch(db, article_id)
    if row is None:
        raise HTTPException(status_code=404, detail="글을 찾을 수 없습니다")
    return row

예상 못 한 예외까지 한 형태로 응답하려면 main.py 에 글로벌 핸들러를 둬요. 스택 트레이스는 로그로만 남기고 응답 본문에는 절대 노출하지 않아요.

# main.py
import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def on_unhandled(request: Request, exc: Exception):
    logger.exception("처리되지 않은 예외")
    return JSONResponse(status_code=500, content={"detail": "서버 오류"})

속도 제한 + CORS

같은 IP 가 엔드포인트를 남용하는 걸 막는 rate limit 은 직접 카운터를 짜기보다 slowapi 같은 도구의 데코레이터로 선언해요.

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.get("/search")
@limiter.limit("10/minute")
async def search(request: Request, q: str):
    ...

데코레이터만으로는 동작하지 않아요 — main.py 에서 limiter 를 앱에 붙이고 초과 예외 핸들러를 등록해야 데코레이터가 요청 시점에 살아나요.

# main.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

브라우저에서 다른 출처로 호출받는 API 는 CORSMiddleware 로 허용 출처를 명시 해요. allow_origins=["*"] 는 자격 증명 요청과 함께 못 쓰고 범위도 지나치게 넓어요 — 도메인을 나열해요.

# main.py
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],   # "*" 금지 — 명시 목록
    allow_methods=["*"],
    allow_headers=["*"],
)

직접 해 보기

articles 리소스 하나를 routers/articles/articles.py 로 만들어 보세요. schemas.py 에 ArticleIn·ArticleOut 을 두고, GET /articles/{id} 가 없는 ID 면 HTTPException(404) 를 던지게 하세요. main.py 에 글로벌 예외 핸들러와 CORSMiddleware 를 등록한 뒤 /docs 에서 직접 호출해 422·404·500 응답을 각각 확인해 보세요.

더 깊이

• FastAPI 의 설계 철학 노트
• Python 백엔드 폴더 철학 노트

강좌 마무리

여덟 단계를 지나왔어요. 왜 Python·FastAPI 인지(1단계)에서 출발해, 폴더를 도메인으로 펼치고(2단계), PostgreSQL 풀에 연결하고(3단계), APScheduler 로 정기 작업을 돌리고(4단계), 크롤러 윤리를 지키며(5단계), 외부 → 가공 → DB 파이프라인을 잇고(6단계), 헬스체크·관측으로 살아 있는지 확인하고(7단계), 마지막으로 그 API 를 실무에서 깨지지 않게 다듬는 패턴(8단계)까지 왔어요.

핵심은 한 줄로 모여요 — Python 한 가지 언어로 데이터를 받아 가공해 저장하고, 그 흐름을 안정적으로 노출하는 전 과정을 직접 만들 수 있다는 것. 작은 단일 파일에서 시작해도 큰 비용 없이 여기까지 자랍니다.

다음 여정이 궁금하면 devops-cloud 에서 이 서비스를 진짜 클라우드에 올리거나, ai-agent-tooling 에서 AI 도구를 다루는 법을 만나 보세요. 여기까지 함께해 주셔서 고맙습니다.