본문으로 건너뛰기
Advertisement

실무 고수 팁

실제 프로덕션 FastAPI 프로젝트에서 적용하는 프로젝트 구조, 예외 처리, 로깅, OpenAPI 커스터마이징 패턴입니다.

프로젝트 구조​

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 앱 진입점
│   ├── config.py            # 환경 설정 (Pydantic Settings)
│   ├── dependencies.py      # 공통 의존성 (DB 세션, 현재 사용자)
│   ├── exceptions.py        # 커스텀 예외
│   ├── models/              # SQLAlchemy ORM 모델
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic 요청/응답 스키마
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── crud/                # DB 접근 레이어
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── routers/             # APIRouter 모음
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   └── utils/               # 유틸리티 함수
│       └── security.py
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── alembic/                 # DB 마이그레이션
├── pyproject.toml
└── .env

환경 설정 (Pydantic Settings)​

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    # 앱 설정
    app_name: str = "My API"
    debug: bool = False
    version: str = "1.0.0"

    # 데이터베이스
    database_url: str = "sqlite+aiosqlite:///./app.db"

    # 보안
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    # 외부 서비스
    redis_url: str = "redis://localhost:6379"

    model_config = {
        "env_file": ".env",
        "env_file_encoding": "utf-8",
    }


@lru_cache
def get_settings() -> Settings:
    return Settings()


# 사용
settings = get_settings()
print(settings.database_url)
# .env 파일
SECRET_KEY=your-super-secret-key-change-me
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/mydb
DEBUG=false

전역 예외 처리​

# exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import logging

logger = logging.getLogger(__name__)


# 커스텀 예외 클래스
class AppException(Exception):
    def __init__(self, status_code: int, detail: str, code: str = "ERROR"):
        self.status_code = status_code
        self.detail = detail
        self.code = code


class NotFoundError(AppException):
    def __init__(self, resource: str, resource_id: int | str):
        super().__init__(
            status_code=404,
            detail=f"{resource} {resource_id}를 찾을 수 없습니다",
            code="NOT_FOUND",
        )


class ConflictError(AppException):
    def __init__(self, detail: str):
        super().__init__(status_code=409, detail=detail, code="CONFLICT")


def setup_exception_handlers(app: FastAPI) -> None:
    # 커스텀 예외 처리
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        logger.warning(f"AppException: {exc.code} - {exc.detail}")
        return JSONResponse(
            status_code=exc.status_code,
            content={"code": exc.code, "detail": exc.detail},
        )

    # Pydantic 검증 실패 — 상세 에러 메시지 형식화
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        errors = [
            {
                "field": ".".join(str(loc) for loc in error["loc"][1:]),
                "message": error["msg"],
            }
            for error in exc.errors()
        ]
        return JSONResponse(
            status_code=422,
            content={"code": "VALIDATION_ERROR", "errors": errors},
        )

    # 500 에러 — 내부 오류 숨기기
    @app.exception_handler(Exception)
    async def general_exception_handler(request: Request, exc: Exception):
        logger.error(f"Unexpected error: {exc}", exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"code": "INTERNAL_ERROR", "detail": "서버 내부 오류가 발생했습니다"},
        )

구조화 로깅​

# logging_config.py
import logging
import json
from datetime import datetime


class JSONFormatter(logging.Formatter):
    """JSON 형식 로그 (로그 수집 시스템 연동용)"""

    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data, ensure_ascii=False)


def setup_logging(debug: bool = False):
    level = logging.DEBUG if debug else logging.INFO

    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())

    logging.basicConfig(
        level=level,
        handlers=[handler],
    )

    # 외부 라이브러리 로그 줄이기
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
# 미들웨어에서 요청 로깅
import logging
from fastapi import Request

logger = logging.getLogger(__name__)


async def log_requests(request: Request, call_next):
    logger.info(
        "Request",
        extra={
            "method": request.method,
            "path": request.url.path,
            "client": request.client.host if request.client else None,
        },
    )
    response = await call_next(request)
    logger.info("Response", extra={"status_code": response.status_code})
    return response

OpenAPI 커스터마이징​

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def custom_openapi(app: FastAPI):
    if app.openapi_schema:
        return app.openapi_schema

    schema = get_openapi(
        title="My Production API",
        version="2.0.0",
        description="""
## 인증
모든 보호된 엔드포인트는 `Authorization: Bearer <token>` 헤더가 필요합니다.

## 에러 코드
- `NOT_FOUND`: 리소스 없음
- `CONFLICT`: 중복 데이터
- `VALIDATION_ERROR`: 입력 검증 실패
        """,
        routes=app.routes,
        tags=[
            {"name": "users", "description": "사용자 관리"},
            {"name": "items", "description": "아이템 관리"},
            {"name": "auth", "description": "인증/인가"},
        ],
    )

    # JWT 보안 스키마 추가
    schema["components"]["securitySchemes"] = {
        "BearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
        }
    }

    app.openapi_schema = schema
    return schema


# main.py
app = FastAPI(docs_url="/docs", redoc_url="/redoc")
app.openapi = lambda: custom_openapi(app)

최종 main.py 조합​

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import get_settings
from exceptions import setup_exception_handlers
from logging_config import setup_logging
from database import create_tables
from routers import users, items, auth

settings = get_settings()
setup_logging(debug=settings.debug)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    yield


app = FastAPI(
    title=settings.app_name,
    version=settings.version,
    lifespan=lifespan,
    docs_url="/docs" if settings.debug else None,  # 프로덕션에서 문서 숨기기
)

# 미들웨어
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 예외 핸들러
setup_exception_handlers(app)

# 라우터
app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])


@app.get("/health")
def health_check():
    return {"status": "ok", "version": settings.version}

정리​

패턴설명
계층 구조routers → crud → models
Pydantic Settings.env 기반 환경 설정
전역 예외 핸들러일관된 에러 응답 형식
JSON 로깅로그 수집 시스템 연동
OpenAPI 커스터마이징문서 개선, 보안 스키마 추가
lifespan앱 시작/종료 이벤트 관리

실무에서는 **관심사 분리(라우터·CRUD·스키마·모델 계층)**와 환경별 설정 분리가 유지보수성의 핵심입니다.

Advertisement
Advertisement