SlowAPI: FastAPI에서 Rate Limiting 구현

# 설치
pip install slowapi
pip install redis  # Redis 백엔드 사용시

기본 Rate Limiting 설정

from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/")
@limiter.limit("5/minute")
async def read_root():
    return {"Hello": "World"}

Redis 백엔드 사용

from slowapi.middleware import SlowAPIMiddleware
from slowapi.storage import RedisStorage
import redis

redis_client = redis.Redis(host='localhost', port=6379)
limiter = Limiter(
    key_func=get_remote_address, 
    storage_uri="redis://localhost:6379"
)

app.add_middleware(SlowAPIMiddleware)

커스텀 키 함수 구현

from fastapi import Request

def get_user_api_key(request: Request):
    api_key = request.headers.get("X-API-Key")
    return api_key or get_remote_address(request)

limiter = Limiter(key_func=get_user_api_key)

@app.get("/api/data")
@limiter.limit("100/day")
async def get_data(request: Request):
    return {"data": "some data"}

동적 제한 설정

def rate_limit_by_user_tier(request: Request):
    user_tier = get_user_tier(request)  # 사용자 등급 확인
    limits = {
        "basic": "10/minute",
        "premium": "100/minute",
        "enterprise": "1000/minute"
    }
    return limits.get(user_tier, "5/minute")

@app.get("/api/premium")
@limiter.limit(rate_limit_by_user_tier)
async def premium_endpoint(request: Request):
    return {"status": "success"}

경로별 제한 설정

@app.get("/api/public")
@limiter.limit("10/minute")
async def public_endpoint(request: Request):
    return {"access": "public"}

@app.get("/api/private")
@limiter.limit("30/minute", key_func=lambda _: "global")
async def private_endpoint(request: Request):
    return {"access": "private"}

@app.post("/api/write")
@limiter.limit("5/minute;100/day")
async def write_endpoint(request: Request):
    return {"operation": "write"}

에러 처리 커스터마이징

from fastapi import HTTPException
from slowapi.errors import RateLimitExceeded

@app.exception_handler(RateLimitExceeded)
async def custom_rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "error": "Rate limit exceeded",
            "reset_at": exc.reset_at.isoformat(),
            "retry_after": exc.retry_after
        }
    )

그룹 제한 설정

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["100/day", "10/minute"]
)

@app.post("/upload")
@limiter.limit(
    "10/hour",
    key_func=lambda r: f"{get_remote_address(r)}/upload"
)
async def upload_file(request: Request):
    return {"status": "uploaded"}

 

주요 기능 및 장점

1. 유연한 설정
   • 다양한 시간 단위 지원
   • 동적 제한 설정
   • 백엔드 선택 가능
2. 확장성
   • Redis 백엔드 지원
   • 커스텀 키 함수
   • 그룹 제한 설정
3. 사용자 경험
   • 상세한 에러 메시지
   • 재시도 정보 제공
   • 헤더 기반 상태 추적
4. 모니터링
   • 제한 상태 추적
   • 사용량 통계
   • 디버깅 용이

SlowAPI를 사용하면 API의 사용량을 효과적으로 제어하고 서버 리소스를 보호할 수 있습니다.