Redis 캐싱으로 FastAPI 응답속도 10배 올리기

Redis 캐싱이란? 왜 FastAPI에 필요한가

FastAPI로 서비스를 운영하다 보면 특정 API 엔드포인트가 반복적으로 동일한 데이터를 데이터베이스에서 조회하는 상황이 생깁니다. 예를 들어 사용자 프로필, 공지사항 목록, 카테고리 트리처럼 변경 빈도가 낮은 데이터를 매 요청마다 PostgreSQL에서 가져오는 것은 엄청난 낭비입니다. Redis는 인메모리 데이터 스토어로, 이런 반복 조회를 밀리초 단위로 처리해 DB 부하를 획기적으로 줄여줍니다.

실제로 Redis를 적용한 뒤 응답 시간이 300ms → 30ms 이하로 줄어드는 경우가 흔합니다. 이 글에서는 FastAPI 프로젝트에 Redis를 연동하고, 캐시 전략(TTL, 무효화, 패턴 삭제)을 설계하는 실전 방법을 코드와 함께 정리합니다.

1. Redis 연결 설정 — aioredis + FastAPI lifespan

FastAPI에서 Redis를 비동기로 사용하려면 redis[asyncio] 패키지가 필요합니다. FastAPI의 lifespan 이벤트를 활용해 앱 시작 시 Redis 연결을 생성하고, 종료 시 정리하는 것이 권장 패턴입니다.

# requirements.txt
redis[asyncio]==5.0.1
fastapi==0.110.0
uvicorn==0.29.0

# app/core/redis.py
import redis.asyncio as aioredis
from contextlib import asynccontextmanager
from fastapi import FastAPI

redis_client: aioredis.Redis | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    redis_client = await aioredis.from_url(
        "redis://localhost:6379",
        encoding="utf-8",
        decode_responses=True,
        max_connections=20,
    )
    yield
    await redis_client.aclose()

def get_redis() -> aioredis.Redis:
    if redis_client is None:
        raise RuntimeError("Redis not initialized")
    return redis_client

# main.py
from app.core.redis import lifespan
app = FastAPI(lifespan=lifespan)

max_connections=20은 커넥션 풀 크기입니다. 트래픽에 따라 조정하세요. decode_responses=True로 설정하면 bytes 대신 str로 자동 디코딩되어 편리합니다.

2. 캐시 데코레이터 패턴 — 재사용 가능한 캐싱 레이어

매 라우터마다 캐시 로직을 반복하면 코드가 지저분해집니다. 범용 캐시 헬퍼 함수를 만들어두면 한 줄로 어떤 데이터든 캐싱할 수 있습니다. Cache-Aside 패턴(Lazy Loading)을 구현합니다: 캐시에 없으면 DB에서 가져와 저장, 있으면 바로 반환.

# app/core/cache.py
import json
import functools
from typing import Any, Callable, Optional
from app.core.redis import get_redis

async def get_cached(
    key: str,
    fetch_fn: Callable,
    ttl: int = 300,  # 기본 5분
    *args,
    **kwargs,
) -> Any:
    """Cache-Aside 패턴 범용 캐시 헬퍼"""
    redis = get_redis()

    # 1. 캐시 확인
    cached = await redis.get(key)
    if cached is not None:
        return json.loads(cached)

    # 2. 캐시 미스 → DB 조회
    data = await fetch_fn(*args, **kwargs)

    # 3. 캐시 저장
    if data is not None:
        await redis.setex(key, ttl, json.dumps(data, ensure_ascii=False, default=str))

    return data

# 사용 예시 — 라우터
from app.core.cache import get_cached

@router.get("/categories")
async def get_categories(db: AsyncSession = Depends(get_db)):
    async def fetch():
        result = await db.execute(select(Category).order_by(Category.order))
        return [c.__dict__ for c in result.scalars().all()]

    return await get_cached("categories:all", fetch, ttl=600)  # 10분

카테고리처럼 거의 바뀌지 않는 데이터는 TTL을 10분~1시간으로 설정해도 무방합니다. 반면 재고 수량처럼 민감한 데이터는 30초~1분으로 짧게 가져가세요.

3. TTL 전략 — 데이터 성격에 따른 만료 시간 설계

캐싱에서 가장 중요한 결정은 TTL(Time To Live) 설정입니다. 너무 길면 오래된 데이터를 서빙하고, 너무 짧으면 캐시 히트율이 떨어져 의미가 없습니다. 데이터 유형별 권장 TTL을 정리했습니다.

• 정적 콘텐츠(공지사항, 약관, 카테고리): 10분~1시간
• 사용자 프로필: 5분 (변경 시 즉시 무효화)
• 검색 결과: 2~5분
• 랭킹/통계: 1분
• 실시간 데이터(재고, 잔액): 캐싱 지양 또는 30초 이하

4. 캐시 무효화 — 데이터 변경 시 즉시 갱신

캐시의 가장 큰 문제는 stale data(오래된 데이터)입니다. 사용자가 프로필을 수정했는데 5분간 이전 데이터가 보이면 안 됩니다. 데이터가 변경될 때 해당 캐시를 즉시 삭제하는 Write-Through 무효화 전략이 필요합니다.

# app/core/cache.py에 무효화 헬퍼 추가
async def invalidate_cache(*keys: str) -> None:
    """단일 또는 복수 캐시 키 무효화"""
    redis = get_redis()
    if keys:
        await redis.delete(*keys)

async def invalidate_pattern(pattern: str) -> int:
    """패턴 매칭으로 캐시 일괄 삭제 (ex: user:123:*)"""
    redis = get_redis()
    keys = await redis.keys(pattern)
    if keys:
        return await redis.delete(*keys)
    return 0

# 사용 예시 — 프로필 업데이트 라우터
@router.put("/users/{user_id}")
async def update_user(
    user_id: int,
    payload: UserUpdateSchema,
    db: AsyncSession = Depends(get_db),
):
    # DB 업데이트
    user = await update_user_in_db(db, user_id, payload)

    # 관련 캐시 즉시 무효화
    await invalidate_cache(
        f"user:{user_id}:profile",
        f"user:{user_id}:summary",
    )
    # 관리자 목록 캐시도 일괄 삭제
    await invalidate_pattern("users:list:*")

    return user

invalidate_pattern은 내부적으로 KEYS 명령을 사용하므로 프로덕션 대규모 환경에서는 SCAN을 사용하는 것이 더 안전합니다. 키가 수만 개 이상이라면 SCAN 기반 구현으로 교체하세요.

5. 성능 측정 — 캐시 적용 전후 벤치마크

캐싱 효과를 수치로 검증하는 것이 중요합니다. httpx와 time을 활용한 간단한 벤치마크로 응답 시간을 비교해보세요. 실제 프로젝트에서 카테고리 조회 API를 캐싱한 결과: 평균 응답 282ms → 8ms (약 35배 개선)를 확인했습니다.

# benchmark.py
import asyncio
import time
import httpx

async def benchmark(url: str, iterations: int = 100):
    async with httpx.AsyncClient() as client:
        times = []
        for _ in range(iterations):
            start = time.perf_counter()
            await client.get(url)
            elapsed = (time.perf_counter() - start) * 1000
            times.append(elapsed)

        avg = sum(times) / len(times)
        p95 = sorted(times)[int(iterations * 0.95)]
        print(f"평균: {avg:.1f}ms | P95: {p95:.1f}ms | 최소: {min(times):.1f}ms")

async def main():
    base = "http://localhost:8000"
    print("=== 캐시 없음 ===")
    await benchmark(f"{base}/categories?cache=0")
    print("=== 캐시 적용 ===")
    await benchmark(f"{base}/categories")

asyncio.run(main())

# 결과 예시
# === 캐시 없음 ===
# 평균: 284.3ms | P95: 412.1ms | 최소: 198.7ms
# === 캐시 적용 ===
# 평균: 7.8ms  | P95: 12.4ms  | 최소: 5.2ms

마무리 — Redis 캐싱 도입 체크리스트

Redis 캐싱을 FastAPI에 도입할 때 기억해야 할 핵심 포인트를 정리합니다.

• 반복 조회가 많고 변경이 드문 데이터부터 캐싱을 시작하세요.
• TTL은 데이터 변경 주기에 맞게 보수적으로 설정하고, 점차 늘려가세요.
• 데이터가 변경될 때 반드시 캐시를 무효화하는 로직을 함께 작성하세요.
• 캐시 키 네이밍을 일관되게 관리하세요 (예: resource:id:field).
• Redis 메모리 한계를 설정하고 eviction policy를 allkeys-lru로 지정하세요.

코드벤터는 FastAPI, SvelteKit, Redis 등 현대적인 기술 스택을 활용해 글로벌 협력 네트워크와 함께 고성능 웹 서비스를 설계하고 개발합니다. 성능 최적화, 캐싱 전략, 아키텍처 설계까지 — 실전에서 검증된 기술을 공유하는 코드벤터 블로그를 구독해보세요.