Omezení četnosti požadavků

Nasazení AI do produkce s FastAPI

Matt Eckerle

Software and Data Engineering Leader

Úvod do omezení četnosti požadavků

semafor

Účel: Řídí frekvenci požadavků na API.
Odpověď: Při překročení limitu vrátí HTTP 429 („Too Many Requests").

Jak omezení četnosti funguje

1. část vývojového diagramu vysvětlujícího omezení četnosti požadavků

Ověřování příchozích přihlašovacích údajů

2. část vývojového diagramu vysvětlujícího omezení četnosti požadavků

Kontrola omezení četnosti

3. část vývojového diagramu vysvětlujícího omezení četnosti požadavků

Nastavení 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"

Logika omezovače četnosti

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. část vývojového diagramu logiky omezení četnosti požadavků

Mazání starých požadavků

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. část vývojového diagramu logiky omezení četnosti požadavků

Kontrola počtu požadavků

    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

Diagram znázorňující kontrolu počtu požadavků API: pokud je počet roven limitu nebo ho překračuje, vrátí true, jinak false.

Přidání kontroly limitu četnosti

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

Použití limitu četnosti na endpoint

@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
    }

Odeslání požadavku 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"}'

Výstup:

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

Pojďme cvičit!

Nasazení AI do produkce s FastAPI