FastAPI 백엔드 API - 고성능 RESTful API 서비스

프로젝트 개요

이 프로젝트는 고성능 RESTful API를 제공하는 FastAPI 기반 백엔드 서비스입니다. 유저 인증, 데이터 CRUD 작업, 캐싱, 그리고 비동기 작업 처리를 위한 확장 가능한 마이크로서비스 아키텍처를 구축했습니다.

주요 목표:

• 초당 10,000개 이상의 요청 처리 (QPS 10K+)
• 99.9% 가용성 (SLA)
• JWT 기반 보안 인증 시스템
• Redis 캐싱으로 응답 시간 50ms 미만 유지

기술 스택

카테고리	기술
프레임워크	FastAPI 0.104+, Python 3.11
데이터베이스	PostgreSQL 15, Redis 7
ORM	SQLAlchemy 2.0, Alembic
인증	JWT (PyJWT), OAuth2.0
비동기 작업	Celery, RabbitMQ
API 문서	OpenAPI 3.0, Swagger UI
테스트	Pytest, Locust (부하 테스트)
배포	Docker, GitHub Actions

주요 기능

1. RESTful API 설계

FastAPI의 자동 API 문서화 기능을 활용하여 OpenAPI 스펙 기반의 API를 설계했습니다.

# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from app.schemas.user import UserResponse, UserCreate
from app.services.user_service import UserService

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),
    user_service: UserService = Depends()
):
    """
    새로운 사용자를 생성합니다.

    - **email**: 유니크한 이메일 주소
    - **password**: 최소 8자 이상
    """
    existing_user = await user_service.get_by_email(db, user_data.email)
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="이미 존재하는 이메일입니다"
        )

    user = await user_service.create(db, user_data)
    return user

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    user_service: UserService = Depends()
):
    """
    특정 사용자 정보를 조회합니다.
    """
    user = await user_service.get(db, user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="사용자를 찾을 수 없습니다"
        )
    return user

2. JWT 기반 인증 시스템

액세스 토큰과 리프레시 토큰을 사용한 보안 인증 시스템을 구현했습니다.

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

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """비밀번호 검증"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """비밀번호 해싱"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """액세스 토큰 생성"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)

    to_encode.update({"exp": expire, "type": "access"})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    """리프레시 토큰 생성"""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

3. Redis 캐싱 레이어

자주 조회되는 데이터를 Redis에 캐싱하여 데이터베이스 부하를 줄였습니다.

# app/services/cache_service.py
from redis.asyncio import Redis
import json
from typing import Optional, Any
from app.core.config import settings

redis_client = Redis.from_url(settings.REDIS_URL, decode_responses=True)

class CacheService:
    def __init__(self, prefix: str = "app", ttl: int = 3600):
        self.prefix = prefix
        self.ttl = ttl

    def _make_key(self, key: str) -> str:
        return f"{self.prefix}:{key}"

    async def get(self, key: str) -> Optional[Any]:
        """캐시 조회"""
        cached = await redis_client.get(self._make_key(key))
        if cached:
            return json.loads(cached)
        return None

    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
        """캐시 저장"""
        try:
            serialized = json.dumps(value)
            await redis_client.setex(
                self._make_key(key),
                ttl or self.ttl,
                serialized
            )
            return True
        except Exception:
            return False

    async def invalidate(self, pattern: str) -> int:
        """패턴 기반 캐시 무효화"""
        keys = []
        async for key in redis_client.scan_iter(match=f"{self.prefix}:{pattern}*"):
            keys.append(key)
        if keys:
            return await redis_client.delete(*keys)
        return 0

4. Celery 비동기 작업 처리

이메일 발송, 데이터 처리 등 무거운 작업을 비동기로 처리합니다.

# app/tasks/email_tasks.py
from celery import Celery
from app.core.config import settings
from app.services.email_service import EmailService

celery_app = Celery(
    "tasks",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_RESULT_BACKEND
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Seoul",
    enable_utc=True,
    task_routes={
        "app.tasks.email_tasks.send_welcome_email": {"queue": "emails"},
        "app.tasks.data_tasks.process_large_dataset": {"queue": "heavy"},
    },
    worker_prefetch_multiplier=1,
)

@celery_app.task(bind=True, max_retries=3)
def send_welcome_email(self, user_email: str, user_name: str):
    """환영 이메일 발송"""
    try:
        email_service = EmailService()
        email_service.send(
            to=user_email,
            subject="환영합니다!",
            template="welcome_email.html",
            context={"user_name": user_name}
        )
        return {"status": "success", "email": user_email}
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

개발 과정

1. 프로젝트 초기 설정 (1주)

• FastAPI 프로젝트 구조 설계 (Clean Architecture 적용)
• PostgreSQL 및 Redis 데이터베이스 스키마 설계
• Alembic 마이그레이션 설정

2. 핵심 API 개발 (2주)

• 유저 인증 및 권한 부여 시스템
• CRUD 기능 구현
• API 문서 자동화 설정

3. 성능 최적화 (2주)

• 쿼리 최적화 및 인덱싱
• Redis 캐싱 레이어 추가
• 비동기 처리 구현

4. 테스트 및 배포 (1주)

• 단위 테스트 및 통합 테스트 작성 (테스트 커버리지 85%)
• Locust를 이용한 부하 테스트
• Docker 컨테이너화 및 CI/CD 파이프라인 구축

문제 해결 과정

문제 1: 높은 동시 요청에서의 데이터베이스 연결 풀 고갈

문제: 동시 요청이 많아질 때 PostgreSQL 연결 풀이 고갈되어 503 에러 발생

원인 분석:

• 연결 풀 사이즈가 너무 작음 (기본값 5)
• 연결이 제대로 반환되지 않는 코드 존재

해결책:

# app/core/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
from app.core.config import settings

# 연결 풀 크기 최적화
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=20,  # 최소 연결 수
    max_overflow=40,  # 추가 연결 수
    pool_timeout=30,  # 연결 대기 시간
    pool_recycle=3600,  # 연결 재생성 주기
    pool_pre_ping=True,  # 연결 유효성 확인
    echo=settings.DEBUG
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

결과: 동시 요청 처리량 1,000 → 5,000으로 증가

문제 2: N+1 쿼리 문제

문제: 관련 데이터를 조회할 때 불필요한 쿼리가 다수 실행

해결책:

# app/repositories/user_repository.py
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from app.models.user import User
from app.models.post import Post

class UserRepository:
    async def get_with_posts(self, db: AsyncSession, user_id: int) -> User:
        """
        관련 포스트를 포함하여 사용자 조회 (Eager Loading)
        """
        result = await db.execute(
            select(User)
            .options(selectinload(User.posts))
            .where(User.id == user_id)
        )
        return result.scalars().first()

결과: 쿼리 수 101개 → 2개로 감소 (98% 개선)

성과와 배운 점

성과

• 성능: 평균 응답 시간 45ms (P99: 200ms)
• 처리량: 초당 12,000 요청 처리 (초과 목표 달성)
• 가용성: 99.95% (월간 다운타임 21분)
• 테스트 커버리지: 85%

배운 점

1. 비동기 프로그래밍: Python async/await 패턴의 중요성
2. 데이터베이스 최적화: 인덱싱과 쿼리 최적화의 파급 효과
3. 캐싱 전략: 캐시 무효화 전략과 TTL 설정의 중요성
4. 마이크로서비스 아키텍처: 서비스 분리가 확장성에 미치는 영향

코드 스니펫

의존성 주입 패턴

# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.security import decode_access_token
from app.repositories.user_repository import UserRepository
from app.models.user import User

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

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    """현재 인증된 사용자 반환"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="인증 정보가 유효하지 않습니다",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = decode_access_token(token)
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user_repo = UserRepository(db)
    user = await user_repo.get(user_id)
    if user is None:
        raise credentials_exception

    return user

API 문서 예시

OpenAPI 스펙을 통해 자동 생성된 API 문서:

GET /api/users/{user_id}

Parameters:
  - user_id (path, integer, required): 사용자 ID

Responses:
  200:
    description: 성공
    schema: UserResponse
  404:
    description: 사용자를 찾을 수 없음
    schema: ErrorResponse
  401:
    description: 인증 실패
    schema: ErrorResponse

향후 개선 계획

• GraphQL 지원
• 이벤트 기반 아키텍처로 마이그레이션
• gRPC 마이크로서비스 간 통신
• 분산 추적 시스템 (Jaeger) 도입
• 머신러닝 기반 이상 탐지

---

GitHub: https://github.com/username/fastapi-backend API Docs: https://api.example.com/docs