라우터 구현과 에러 처리 - Part 2

라우터 구조화

대규모 애플리케이션에서는 라우터를 체계적으로 구성하는 것이 중요합니다. 버전별, 기능별로 라우터를 분리하여 관리합니다.

# app/api/v1/router.py
from fastapi import APIRouter
from app.api.v1.endpoints import user, item

api_router = APIRouter()

api_router.include_router(
    user.router,
    prefix="/users",
    tags=["users"]
)

api_router.include_router(
    item.router,
    prefix="/items",
    tags=["items"]
)

# app/api/v1/endpoints/user.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.api import deps
from app.services import user as user_service
from app.schemas.user import UserCreate, UserInDB

router = APIRouter()

@router.post("/", response_model=UserInDB)
async def create_user(
    *,
    db: AsyncSession = Depends(deps.get_db),
    user_in: UserCreate,
) -> UserInDB:
    user = await user_service.get_user_by_email(db, user_in.email)
    if user:
        raise HTTPException(
            status_code=400,
            detail="Email already registered"
        )
    return await user_service.create_user(db, user_in)

예외 처리와 커스텀 예외

일관된 에러 응답을 위한 예외 처리 시스템을 구현합니다.

# app/core/exceptions.py
from fastapi import HTTPException, status
from typing import Any

class AppException(HTTPException):
    def __init__(
        self,
        status_code: int,
        detail: Any = None,
        headers: dict = None
    ) -> None:
        super().__init__(status_code=status_code, detail=detail, headers=headers)

class NotFoundException(AppException):
    def __init__(self, detail: str = "Resource not found") -> None:
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=detail
        )

class BadRequestException(AppException):
    def __init__(self, detail: str = "Bad request") -> None:
        super().__init__(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=detail
        )

# app/api/v1/endpoints/user.py
@router.get("/{user_id}", response_model=UserInDB)
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(deps.get_db)
) -> UserInDB:
    user = await user_service.get_user(db, user_id)
    if not user:
        raise NotFoundException(f"User {user_id} not found")
    return user

미들웨어 구현

애플리케이션 전반에 걸쳐 적용되는 미들웨어를 구현합니다.

# app/core/middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
from app.core.logger import logger

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time

        logger.info(
            f"Path: {request.url.path} "
            f"Method: {request.method} "
            f"Processing Time: {process_time:.3f}s"
        )

        return response

# app/main.py
from fastapi import FastAPI
from app.core.middleware import LoggingMiddleware
from app.api.v1.router import api_router
from app.core.config import settings

app = FastAPI(title=settings.PROJECT_NAME)
app.add_middleware(LoggingMiddleware)
app.include_router(api_router, prefix=settings.API_V1_STR)

페이지네이션 구현

재사용 가능한 페이지네이션 컴포넌트를 구현합니다.

# app/core/pagination.py
from typing import Generic, TypeVar, Sequence
from pydantic import BaseModel
from fastapi import Query

T = TypeVar("T")

class PageParams:
    def __init__(
        self,
        page: int = Query(1, ge=1, description="Page number"),
        size: int = Query(10, ge=1, le=100, description="Items per page")
    ):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

class Page(BaseModel, Generic[T]):
    items: Sequence[T]
    total: int
    page: int
    size: int
    pages: int

    @classmethod
    def create(cls, items: Sequence[T], total: int, params: PageParams) -> "Page[T]":
        pages = -(-total // params.size)  # 올림 나눗셈
        return cls(
            items=items,
            total=total,
            page=params.page,
            size=params.size,
            pages=pages
        )

# 사용 예시
@router.get("/", response_model=Page[UserInDB])
async def get_users(
    pagination: PageParams = Depends(),
    db: AsyncSession = Depends(deps.get_db)
) -> Page[UserInDB]:
    users, total = await user_service.get_users_paginated(
        db,
        skip=pagination.offset,
        limit=pagination.size
    )
    return Page.create(users, total, pagination)

캐싱 구현

성능 최적화를 위한 캐싱 시스템을 구현합니다.

# app/core/cache.py
from functools import wraps
from redis import asyncio as aioredis
import json
from app.core.config import settings

redis = aioredis.from_url(settings.REDIS_URL)

def cache(expire: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 캐시 키 생성
            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"

            # 캐시된 데이터 확인
            cached = await redis.get(key)
            if cached:
                return json.loads(cached)

            # 새 데이터 가져오기
            result = await func(*args, **kwargs)

            # 캐시 저장
            await redis.set(
                key,
                json.dumps(result),
                ex=expire
            )

            return result
        return wrapper
    return decorator

# 사용 예시
@router.get("/{user_id}", response_model=UserInDB)
@cache(expire=300)  # 5분 캐시
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(deps.get_db)
) -> UserInDB:
    user = await user_service.get_user(db, user_id)
    if not user:
        raise NotFoundException(f"User {user_id} not found")
    return user

이상으로 Part 2에서는 FastAPI의 라우터 구현, 예외 처리, 미들웨어, 페이지네이션, 캐싱 등 실전적인 내용을 다뤄보았습니다. Part 3에서는 테스트 작성, 배포 전략, 성능 최적화 등에 대해 다루도록 하겠습니다.