테스트, 배포, 성능 최적화 - Part 3

테스트 구현

테스트는 애플리케이션의 안정성을 보장하는 핵심입니다. pytest를 활용하여 체계적인 테스트를 구현해보겠습니다.

# tests/conftest.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.api import deps
from app.db.base import Base

TEST_DATABASE_URL = "postgresql+asyncpg://test:test@localhost/test_db"

@pytest.fixture(scope="session")
async def test_engine():
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def db_session(test_engine):
    async_session = sessionmaker(
        test_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        yield session

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[deps.get_db] = override_get_db
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# tests/api/test_users.py
import pytest
from app.schemas.user import UserCreate

async def test_create_user(client: AsyncClient):
    user_data = {
        "email": "test@example.com",
        "password": "testpassword"
    }
    response = await client.post("/api/v1/users/", json=user_data)
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == user_data["email"]
    assert "id" in data

성능 최적화

대규모 애플리케이션에서 성능은 매우 중요합니다. 주요 최적화 전략들을 살펴보겠습니다.

# app/core/optimizations.py
from functools import wraps
from sqlalchemy import select
from sqlalchemy.orm import selectinload

def optimize_query(model, *related_models):
    def decorator(func):
        @wraps(func)
        async def wrapper(db: AsyncSession, *args, **kwargs):
            query = select(model)
            for related in related_models:
                query = query.options(selectinload(related))
            result = await db.execute(query)
            return result.scalars().all()
        return wrapper
    return decorator

# app/services/user.py
@optimize_query(User, User.items)
async def get_users_with_items(db: AsyncSession):
    pass  # 쿼리는 데코레이터에서 처리됨

# Background Tasks 활용
from fastapi import BackgroundTasks

@router.post("/users/")
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(deps.get_db)
):
    user = await user_service.create_user(db, user_in)
    background_tasks.add_task(send_welcome_email, user.email)
    return user

보안 설정

애플리케이션 보안을 강화하기 위한 설정들입니다.

# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(
        to_encode,
        settings.SECRET_KEY,
        algorithm=settings.ALGORITHM
    )
    return encoded_jwt

# 미들웨어로 Rate Limiting 구현
from fastapi import Request
import time
from app.core.cache import redis

class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        key = f"rate_limit:{client_ip}"

        # 1분당 최대 60회 요청 제한
        requests = await redis.incr(key)
        if requests == 1:
            await redis.expire(key, 60)

        if requests > 60:
            raise HTTPException(
                status_code=429,
                detail="Too many requests"
            )

        return await call_next(request)

배포 설정

Docker를 활용한 배포 설정입니다.

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      - db
      - redis

  db:
    image: postgres:13
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      - POSTGRES_DB=fastapi
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password

  redis:
    image: redis:6
    ports:
      - "6379:6379"

volumes:
  postgres_data:

모니터링 설정

애플리케이션 모니터링을 위한 설정입니다.

# app/core/monitoring.py
from prometheus_client import Counter, Histogram
import time

REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint']
)

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

        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=request.url.path,
            status=response.status_code
        ).inc()

        REQUEST_LATENCY.labels(
            method=request.method,
            endpoint=request.url.path
        ).observe(duration)

        return response

이상으로 FastAPI 시리즈를 마무리하겠습니다. 이 시리즈에서는 프로젝트 구조부터 테스트, 배포, 성능 최적화까지 FastAPI를 사용한 대규모 애플리케이션 개발에 필요한 다양한 측면을 다뤄보았습니다. 각 부분은 프로젝트의 요구사항에 맞게 조정하여 사용하시면 됩니다.