FastAPI-Cache로 구현하는 효율적인 API 캐싱

FastAPI-Cache 소개와 설치

FastAPI-Cache는 FastAPI 애플리케이션에서 손쉽게 캐싱을 구현할 수 있게 해주는 라이브러리입니다. Redis, Memcached 등 다양한 백엔드를 지원합니다.

pip install fastapi-cache2[redis]  # Redis 백엔드 사용시

기본 설정

FastAPI 애플리케이션에 캐시를 설정하는 방법입니다.

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from redis import asyncio as aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

기본적인 캐시 사용

엔드포인트에 캐시를 적용하는 가장 기본적인 방법입니다.

from fastapi_cache.decorator import cache

@app.get("/users/{user_id}")
@cache(expire=60)  # 60초 동안 캐시
async def get_user(user_id: int):
    # 데이터베이스 조회 시뮬레이션
    await asyncio.sleep(1)  # 실제 DB 조회라고 가정
    return {"id": user_id, "name": f"User {user_id}"}

# 동적 만료 시간 설정
def get_cache_expire(request: Request):
    if "premium" in request.headers:
        return 30  # premium 사용자는 30초
    return 60     # 일반 사용자는 60초

@app.get("/products/{product_id}")
@cache(expire=get_cache_expire)
async def get_product(product_id: int):
    return {"id": product_id, "name": f"Product {product_id}"}

캐시 키 커스터마이징

캐시 키를 커스터마이징하여 더 세밀한 캐시 제어가 가능합니다.

from fastapi_cache.decorator import cache
from fastapi import Request

def custom_key_builder(
    func,
    namespace: str | None = "",
    request: Request | None = None,
    *args,
    **kwargs,
):
    prefix = FastAPICache.get_prefix()
    cache_key = f"{prefix}:{namespace}:{func.__module__}:{func.__name__}:"

    # URL 파라미터 포함
    if kwargs:
        cache_key += ":".join(f"{k}:{v}" for k, v in kwargs.items())

    # 쿼리 파라미터 포함
    if request:
        query_params = request.query_params
        if query_params:
            cache_key += ":".join(f"{k}:{v}" for k, v in query_params.items())

    return cache_key

@app.get("/search")
@cache(expire=300, key_builder=custom_key_builder)
async def search_items(q: str, category: str | None = None):
    # 검색 로직
    return {"query": q, "category": category}

조건부 캐싱

특정 조건에서만 캐시를 적용하는 방법입니다.

from typing import Callable

def cache_if(condition: Callable[..., bool]):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if not condition(*args, **kwargs):
                return await func(*args, **kwargs)

            # 캐시 로직
            cache_key = f"{func.__module__}:{func.__name__}:{args}:{kwargs}"
            cached_value = await FastAPICache.get(cache_key)

            if cached_value is not None:
                return cached_value

            value = await func(*args, **kwargs)
            await FastAPICache.set(cache_key, value, expire=60)
            return value

        return wrapper
    return decorator

def should_cache(user_id: int) -> bool:
    return user_id > 100  # ID가 100보다 큰 사용자만 캐시

@app.get("/users/{user_id}/stats")
@cache_if(should_cache)
async def get_user_stats(user_id: int):
    return {"id": user_id, "stats": "some heavy computation"}

캐시 무효화

캐시를 수동으로 무효화하는 방법입니다.

from fastapi_cache.decorator import cache

@app.post("/users")
async def create_user(user: UserCreate):
    new_user = await create_user_in_db(user)
    # users 네임스페이스의 모든 캐시 삭제
    await FastAPICache.clear(namespace="users")
    return new_user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    await delete_user_from_db(user_id)
    # 특정 사용자 관련 캐시만 삭제
    cache_key = f"user:{user_id}"
    await FastAPICache.delete(cache_key)
    return {"status": "success"}

고급 캐시 패턴 구현

더 복잡한 캐싱 시나리오를 위한 구현입니다.

from fastapi_cache.decorator import cache
from typing import Optional

class CacheSettings:
    def __init__(
        self,
        expire: int = 60,
        namespace: Optional[str] = None,
        key_builder: Optional[callable] = None
    ):
        self.expire = expire
        self.namespace = namespace
        self.key_builder = key_builder

class CachedAPIRouter:
    def __init__(self, cache_settings: CacheSettings):
        self.router = APIRouter()
        self.cache_settings = cache_settings

    def cached_route(self, path: str, **kwargs):
        def decorator(func):
            # 캐시 데코레이터 적용
            cached_func = cache(
                expire=self.cache_settings.expire,
                namespace=self.cache_settings.namespace,
                key_builder=self.cache_settings.key_builder
            )(func)

            # 라우터에 등록
            self.router.add_api_route(
                path,
                cached_func,
                **kwargs
            )
            return cached_func
        return decorator

# 사용 예시
user_cache_settings = CacheSettings(
    expire=300,
    namespace="users",
    key_builder=custom_key_builder
)

user_router = CachedAPIRouter(user_cache_settings)

@user_router.cached_route("/users/{user_id}", methods=["GET"])
async def get_user(user_id: int):
    return {"id": user_id}

app.include_router(user_router.router)

FastAPI-Cache를 활용하면 API의 성능을 크게 향상시킬 수 있습니다. 특히 데이터베이스 조회가 빈번하거나 계산 비용이 높은 엔드포인트에서 효과적입니다. 단, 캐시 만료 시간과 무효화 전략은 비즈니스 요구사항에 맞게 신중히 설정해야 합니다.