[FastAPI] FastAPI 속도개선 - 캐시(Cache) (15)

 

 

1. 캐시(Cache)란

• 캐시(Cache)는 자주 사용되는 데이터를 빠르게 접근할 수 있도록 저장해 두는 임시 저장소
• 즉, 속도를 높이고 성능을 최적화하기 위해 데이터를 미리 저장하는 기술

 

동작 방식

• 사용자가 특정 데이터 요청 → 서버에서 데이터 조회 → 캐시에 저장
• 다음에 동일한 요청이 오면, 데이터베이스 대신 캐시에서 데이터를 반환
• 일정 시간이 지나면 캐시 만료(Expire) → 새로운 데이터를 다시 저장

 

사용 이유

• 속도 향상
• 서버 부하 감소
• 비용 절감

 

주의사항

• 오래된 데이터(Invalidation 문제) : 최신화된 데이터가 불러와지지 않을 수 있음
• 메모리 사용량 증가 : 캐시가 많아질수록 메모리를 많이 차지하므로 적절한 용량 관리가 필요.

 

---

2. FastAPI 사용 방법

2.1. 설치

먼저 아래 명령어로 라이브러리를 설치한다.

pip install fastapi-cache2

 

---

2.2. 기본 사용법

FastAPI에서는 @cache 데코레이터를 사용할 수 있다.

• FastAPI 실행시, FastAPIcache.init(InMemoryBackend()) 로 캐시 데이터를 초기화 한다.
• @cache(expire=10) 데코레이터를 사용하면, 해당 엔드포인트의 결과가 10초 동안 저장됨
• 10초 내에 동일한 요청이 오면 캐싱된 값을 반환하여 데이터베이스 조회나 계산을 생략

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.inmemory import InMemoryBackend
from fastapi_cache.decorator import cache

import time

app = FastAPI()

# FastAPI 시작 시 캐시 백엔드 초기화
@app.on_event("startup")
async def startup():
    FastAPICache.init(InMemoryBackend())

# 10초 동안 동일한 요청 결과를 캐싱
@app.get("/time")
@cache(expire=10)
async def get_time():
    return {"time": time.time()}

 

 

---

2.2. 권장 사용법

In-Memory 캐시는 서버가 재시작되면 사라지므로, 보통 Redis를 캐시 백엔드로 사용한다.

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import aioredis
import time

app = FastAPI()

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

@app.get("/time")
@cache(expire=10)
async def get_time():
    return {"time": time.time()}

 

---

2.3. 캐시 삭제

FastAPICache.clear() 사용

아래  /clear_cache 엔드포인트를 호출하면 모든 캐시가 삭제된다.

from fastapi import Depends
from fastapi_cache import FastAPICache

@app.post("/clear_cache")
async def clear_cache():
    await FastAPICache.clear()
    return {"message": "Cache cleared"}

 

---

3. 사용자별 캐시 적용

각 사용자별로 다른 응답을 캐싱하려면 캐시 키를 사용자별로 다르게 설정해야 함

 

3.1. 로그인한 사용자 ID에 따라 다른 캐시 적용

@cache(expire=10, key_builder=key_builder) 

캐시키 생성 함수인 key_builder 함수를 만들어서 데코레이터에 넣어 줘야함.

key_builder 함수 return은 키값

from fastapi import FastAPI, Depends
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import aioredis

app = FastAPI()

# Redis 캐시 초기화
@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost")
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

# 사용자 인증 시뮬레이션 (예제용)
async def get_current_user():
    return {"user_id": 123}  # 로그인한 사용자 ID (실제 앱에서는 JWT 토큰 등을 활용)

# 사용자별 캐시 키 생성 함수
def custom_key_builder(func, *args, **kwargs):
    user = kwargs.get("user")  # user_id 가져오기
    return f"{func.__name__}:{user['user_id']}"  # user_id별로 캐시 키 생성

@app.get("/user_data")
@cache(expire=10, key_builder=custom_key_builder)  # 사용자별 캐싱
async def get_user_data(user: dict = Depends(get_current_user)):
    return {"user_id": user["user_id"], "data": "This is user-specific data"}

 

---

3.2. 클라이언트별(IP 기준) 캐싱

마찬가지 방법으로 ip 별로 키 생성

from fastapi import Request

# IP 기반 캐시 키 생성 함수
def ip_based_key_builder(func, request: Request, *args, **kwargs):
    client_ip = request.client.host  # 요청한 클라이언트 IP 가져오기
    return f"{func.__name__}:ip:{client_ip}"

@app.get("/client_specific_data")
@cache(expire=10, key_builder=ip_based_key_builder)
async def client_data(request: Request):
    return {"client_ip": request.client.host, "data": "This data is specific to your IP"}