FastAPI 실전 프로덕션 가이드 — 비동기 패턴, 인증, Docker 배포까지

프로젝트 디렉토리 구조
app/
├── main.py              # FastAPI 앱 인스턴스, 라우터 등록
├── config.py            # 환경변수 (pydantic-settings)
├── database.py          # SQLAlchemy async 엔진·세션
├── dependencies.py      # 공통 의존성 (현재 유저, DB 세션)
├── models/              # SQLAlchemy ORM 모델
│   ├── user.py
│   └── item.py
├── schemas/             # Pydantic 요청/응답 스키마
│   ├── user.py
│   └── item.py
├── routers/             # 기능별 APIRouter
│   ├── auth.py
│   ├── users.py
│   └── items.py
├── services/            # 비즈니스 로직 (DB 직접 접근 금지)
│   ├── auth_service.py
│   └── item_service.py
├── middleware/          # 커스텀 미들웨어
│   └── logging.py
└── tests/
    ├── conftest.py
    └── test_items.py

app/main.py — 앱 팩토리 패턴
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import auth, users, items
from app.database import engine, Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 앱 시작 시: DB 테이블 생성, 커넥션 풀 초기화
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 앱 종료 시: 커넥션 풀 정리
    await engine.dispose()

app = FastAPI(
    title="My API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router,  prefix="/api/v1/auth",  tags=["auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
app.include_router(items.router, prefix="/api/v1/items", tags=["items"])

app/database.py — async SQLAlchemy 설정
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.config import settings

# asyncpg 드라이버 사용: postgresql+asyncpg://
engine = create_async_engine(
    settings.DATABASE_URL,          # postgresql+asyncpg://user:pass@host/db
    pool_size=10,                   # 커넥션 풀 최대 10개
    max_overflow=20,                # 풀 초과 시 최대 20개 추가
    pool_timeout=30,                # 커넥션 획득 대기 30초
    pool_recycle=1800,              # 30분 후 커넥션 재생성 (RDS 권장)
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    engine, expire_on_commit=False
)

class Base(DeclarativeBase):
    pass

# 의존성 주입용 세션 제공자
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app/schemas/user.py — 요청/응답 스키마 분리
from pydantic import BaseModel, EmailStr, field_validator
from datetime import datetime
from typing import Optional

# 생성 요청 스키마 (비밀번호 포함)
class UserCreate(BaseModel):
    email: EmailStr
    password: str
    name: str

    @field_validator('password')
    @classmethod
    def password_strength(cls, v):
        if len(v) < 8:
            raise ValueError('비밀번호는 최소 8자 이상이어야 합니다')
        return v

# 응답 스키마 (비밀번호 제외)
class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    created_at: datetime

    model_config = {"from_attributes": True}  # ORM 모드 (v2)

# 업데이트 요청 (Optional 필드)
class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None

app/services/auth_service.py — JWT 핵심 로직
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.config import settings

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

def create_access_token(data: dict) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    return jwt.encode(
        {**data, "exp": expire, "type": "access"},
        settings.SECRET_KEY,
        algorithm="HS256"
    )

def create_refresh_token(data: dict) -> str:
    expire = datetime.now(timezone.utc) + timedelta(days=7)
    return jwt.encode(
        {**data, "exp": expire, "type": "refresh"},
        settings.SECRET_KEY,
        algorithm="HS256"
    )

async def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="인증 정보가 유효하지 않습니다",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        if payload.get("type") != "access":
            raise credentials_exception
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = await get_user_by_id(db, user_id)
    if user is None:
        raise credentials_exception
    return user

gunicorn.conf.py — 프로덕션 Gunicorn 설정
import multiprocessing

# 워커 수: CPU 코어 × 2 + 1 (I/O 중심 서비스)
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"

bind = "0.0.0.0:8000"
timeout = 120              # 요청 타임아웃 (초)
keepalive = 5              # keep-alive 연결 유지 시간
max_requests = 1000        # 워커당 최대 요청 수 (메모리 누수 방지)
max_requests_jitter = 100  # 재시작 시점 분산 (동시 재시작 방지)

# 로깅
accesslog = "-"            # stdout
errorlog  = "-"            # stderr
loglevel  = "warning"

# 시작 명령
# gunicorn app.main:app -c gunicorn.conf.py

Dockerfile — 멀티스테이지 빌드
# ── 빌드 스테이지 ──
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# ── 실행 스테이지 ──
FROM python:3.12-slim
WORKDIR /app

# 비루트 사용자 (보안)
RUN addgroup --system appgroup && adduser --system --ingroup appgroup appuser

COPY --from=builder /install /usr/local
COPY . .

RUN chown -R appuser:appgroup /app
USER appuser

EXPOSE 8000
CMD ["gunicorn", "app.main:app", "-c", "gunicorn.conf.py"]

docker-compose.yml — FastAPI + Nginx + PostgreSQL
version: "3.9"
services:
  api:
    build: .
    env_file: .env
    restart: unless-stopped
    depends_on:
      db:
        condition: service_healthy

  nginx:
    image: nginx:1.25-alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
      - ./certbot/conf:/etc/letsencrypt
    depends_on: [api]
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: ${DB_NAME}
      POSTGRES_USER: ${DB_USER}
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pg_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${DB_USER}"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pg_data:

nginx.conf — 리버스 프록시 + 보안 헤더
upstream api {
    server api:8000;
}

server {
    listen 443 ssl;
    server_name api.yourdomain.com;

    ssl_certificate     /etc/letsencrypt/live/api.yourdomain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.yourdomain.com/privkey.pem;

    # 보안 헤더
    add_header X-Frame-Options "DENY";
    add_header X-Content-Type-Options "nosniff";
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains";

    # 요청 크기 제한 (파일 업로드 고려)
    client_max_body_size 10M;

    location / {
        proxy_pass         http://api;
        proxy_set_header   Host $host;
        proxy_set_header   X-Real-IP $remote_addr;
        proxy_set_header   X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header   X-Forwarded-Proto $scheme;
        proxy_read_timeout 90;
    }
}

# HTTP → HTTPS 리다이렉트
server {
    listen 80;
    server_name api.yourdomain.com;
    return 301 https://$host$request_uri;
}

app/worker.py — Celery 설정
from celery import Celery
from app.config import settings

celery_app = Celery(
    "worker",
    broker=settings.REDIS_URL,    # redis://localhost:6379/0
    backend=settings.REDIS_URL,
)

celery_app.conf.update(
    task_serializer="json",
    result_expires=3600,           # 결과 1시간 보관
    worker_prefetch_multiplier=1,  # 한 번에 1개 작업만 가져옴
    task_acks_late=True,           # 작업 완료 후 ACK (크래시 대비)
)

@celery_app.task(bind=True, max_retries=3)
def send_welcome_email(self, user_id: int, email: str):
    try:
        # 이메일 발송 로직
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # 60초 후 재시도

FastAPI 엔드포인트에서 Celery 태스크 발행
from fastapi import APIRouter, Depends
from app.worker import send_welcome_email
from app.schemas.user import UserCreate, UserResponse

router = APIRouter()

@router.post("/register", response_model=UserResponse, status_code=201)
async def register(user_data: UserCreate, db=Depends(get_db)):
    user = await create_user(db, user_data)
    # 비동기 작업 큐에 추가 — 응답 블로킹 없음
    send_welcome_email.delay(user.id, user.email)
    return user

Prometheus 메트릭 엔드포인트 추가
from prometheus_fastapi_instrumentator import Instrumentator

# main.py lifespan 또는 앱 초기화 시
Instrumentator().instrument(app).expose(app)
# → GET /metrics 엔드포인트 자동 생성

Sentry 연동 (3줄)
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn=settings.SENTRY_DSN,
    integrations=[FastApiIntegration()],
    traces_sample_rate=0.2,   # 20% 트랜잭션 추적
)