효율적은 API 응답을 위하여 캐시를 사용하는 방법을 알아보았습니다.
클라이언트 측에서 캐시를 효과적으로 사용하려면 HTTP 캐싱을 통해 브라우저나 프록시 서버가 어떻게 응답을 캐싱할지 지시할 수 있습니다. Cache-Control
헤더를 사용하면 캐싱 정책을 명확히 전달할 수 있습니다.
Cache-Control
헤더는 클라이언트에게 응답을 얼마나 오랫동안, 어떤 조건으로 캐싱할지 지시하는 데 사용됩니다. 대표적인 지시자는 다음과 같습니다:
max-age
: 응답을 캐싱할 수 있는 최대 시간을 초 단위로 설정합니다.no-cache
: 캐시된 응답을 사용하기 전에 서버에 유효성 검사를 요청하게 합니다.no-store
: 응답을 전혀 캐싱하지 않습니다.public
: 응답을 모든 캐시에서 저장할 수 있도록 허용합니다.private
: 특정 사용자만 캐시할 수 있게 합니다.FastAPI에서는 Response
객체를 사용해 쉽게 Cache-Control
헤더를 설정할 수 있습니다. 예를 들어, 다음 코드를 통해 응답에 캐시 헤더를 추가할 수 있습니다.
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/items/")
def read_items(response: Response):
response.headers["Cache-Control"] = "public, max-age=3600"
return {"item": "This is a cached item"}
위 코드에서는 Cache-Control: public, max-age=3600
헤더를 설정하여 클라이언트가 응답을 1시간 동안 캐싱할 수 있게 합니다. 이러한 설정은 정적 데이터나 자주 변경되지 않는 데이터를 캐싱할 때 매우 유용합니다.
서버 측에서도 여러 가지 방식으로 캐싱을 적용할 수 있습니다. FastAPI에서 흔히 사용하는 방법은 In-Memory 캐싱과 Redis를 이용한 캐싱입니다.
In-Memory 캐싱은 서버 메모리에 데이터를 저장하여, 동일한 요청이 들어올 때 빠르게 응답할 수 있도록 합니다. 가장 간단한 방법으로 Python의 functools.lru_cache
를 사용하여 캐싱을 구현할 수 있습니다.
functools.lru_cache
사용하기lru_cache
는 Python 표준 라이브러리에서 제공하는 데코레이터로, 함수의 결과를 캐싱하고 동일한 인자로 호출될 경우 캐시된 결과를 반환합니다.
from fastapi import FastAPI
from functools import lru_cache
app = FastAPI()
@lru_cache(maxsize=32)
def get_data(param: int):
return {"data": param * 2}
@app.get("/compute/{param}")
def compute(param: int):
return get_data(param)
이 코드에서 get_data
함수는 인자로 받은 param
값을 두 배로 처리하고 결과를 캐싱합니다. maxsize
는 캐시할 수 있는 항목의 최대 수를 나타내며, 초과 시 가장 오래된 항목이 제거됩니다. 이 방법은 간단한 캐싱에는 적합하지만, 서버 메모리 사용량이 증가할 수 있어 주의가 필요합니다.
In-Memory 캐싱은 서버 재시작 시 캐시가 사라지거나 여러 서버 인스턴스 간 캐시를 공유하지 못하는 문제가 있습니다. 이를 해결하기 위해 Redis와 같은 외부 캐시 저장소를 사용할 수 있습니다.
Redis는 키-값 구조의 데이터를 인메모리에 저장하는 데이터베이스로, 매우 빠른 읽기/쓰기 성능을 자랑합니다. FastAPI와 함께 Redis를 사용하면 분산 환경에서도 안정적인 캐싱을 구현할 수 있습니다.
Redis를 FastAPI와 함께 사용하려면 비동기 Redis 클라이언트인 aioredis
를 설치하고 사용하는 방법을 소개하겠습니다.
aioredis
설치:
pip install aioredis[fastapi]
Redis와 FastAPI를 연결하고 캐싱을 구현하는 코드:
from fastapi import FastAPI, Depends
from aioredis import from_url
import json
app = FastAPI()
redis = from_url("redis://localhost", decode_responses=True)
async def get_redis():
return redis
@app.get("/items/{item_id}")
async def read_item(item_id: str, redis=Depends(get_redis)):
cached_item = await redis.get(item_id)
if cached_item:
return json.loads(cached_item)
item = {"item_id": item_id, "value": "This is a dynamic value"}
# 캐시에 10분간 저장
await redis.set(item_id, json.dumps(item), ex=600)
return item
위 코드에서는 FastAPI가 Redis에 연결되어, 먼저 캐시된 데이터를 찾고 없을 경우 데이터를 생성하여 캐시에 저장합니다. Redis는 특히 분산 환경에서 안정적으로 캐시를 공유하고 관리할 수 있는 장점이 있습니다.