속도 제한

FastAPI로 AI 프로덕션 배포하기

Matt Eckerle

Software and Data Engineering Leader

속도 제한 소개

신호등

목적: API 요청 빈도를 제어합니다.
응답: 한도 초과 시 HTTP 429("Too Many Requests")를 반환합니다.

속도 제한 작동 원리

속도 제한 로직 흐름도 1부

수신 자격 증명 인증

속도 제한 로직 흐름도 2부

속도 제한 검사

속도 제한 로직 흐름도 3부

API 설정

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import APIKeyHeader
from pydantic import BaseModel

app = FastAPI()
model = SentimentAnalyzer(pkl_file_path)

API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
API_KEY = "your-secret-key"

속도 제한기 로직

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, requests_per_min: int = 10):
        self.requests_per_min = requests_per_min
        self.requests = defaultdict(list)

    def is_rate_limited(
        self, api_key: str
    ) -> tuple[bool, int]:

속도 제한 구현 로직 흐름도 1부

오래된 요청 삭제

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, requests_per_min: int = 10):
        self.requests_per_min = requests_per_min
        self.requests = defaultdict(list)

    def is_rate_limited(
        self, api_key: str
    ) -> tuple[bool, int]:

        now = datetime.now()
        minute_ago = now - timedelta(minutes=1)
        self.requests[api_key] = [
            req_time for req_time in 
          self.requests[api_key]
          if req_time > minute_ago
        ]

속도 제한 로직 흐름도 2부

요청 수 확인

    def is_rate_limited(self, api_key: str) -> 
  tuple[bool, int]:
        now = datetime.now()
        minute_ago = now - timedelta(minutes=1)

        self.requests[api_key] = [
          req_time for req_time in 
          self.requests[api_key]
          if req_time > minute_ago
        ]

        recent_requests = len(self.requests[api_key])
        if recent_requests >= self.requests_per_min:
            return True, 0

        self.requests[api_key].append(now)
        return False

API 요청 수 확인 다이어그램: 요청 수가 한도 이상이면 true, 아니면 false 반환

속도 제한 검사 추가

rate_limiter = RateLimiter(requests_per_minute=10)

def test_api_key(api_key: str = Depends(API_KEY_HEADER)):
    if api_key != API_KEY:
        raise HTTPException(
            status_code=403,
            detail="Invalid API key"
        )
    is_limited, _ = rate_limiter.is_rate_limited(api_key)
    if is_limited:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Please try again later."
        )
    return api_key

엔드포인트에 속도 제한 적용

@app.post("/predict")
def predict_sentiment(
    request: SentimentRequest,
    api_key: str = Depends(test_api_key)
):
    result = sentiment_model(request.text)

    _, requests_remaining = 
           rate_limiter.is_rate_limited(api_key)

    return {
        "text": request.text,
        "sentiment": result[0]["label"].lower(),
        "confidence": result[0]["score"],
        "requests_remaining": requests_remaining
    }

요청 11회 전송:

curl -X POST "http://localhost:8000/predict" \
     -H "Content-Type: application/json" \
     -H "X-API-Key: your-secret-key" \
     -d '{"text": "I love this product"}'

출력:

{"detail":"Rate limit exceeded. 
           Please try again later."}

연습해 봅시다!

FastAPI로 AI 프로덕션 배포하기