Oran Sınırlama

FastAPI ile Üretime AI Dağıtımı

Matt Eckerle

Software and Data Engineering Leader

Oran sınırlamaya giriş

bir trafik lambası

Amaç: API isteklerinin sıklığını kontrol eder.
Yanıt: Limit aşılırsa HTTP 429 ("Too Many Requests") döner.

Oran sınırlama nasıl çalışır

Oran sınırlamayı açıklayan akış şemasının Bölüm 1'i

Gelen kimlik bilgilerinin doğrulanması

Oran sınırlamayı açıklayan akış şemasının Bölüm 2'si

Oran sınırlama kontrolü

Oran sınırlamayı açıklayan akış şemasının Bölüm 3'ü

API'mizi kurmak

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"

Oran sınırlama mantığı

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]:

Oran sınırlama uygulama mantığının akış şemasının Bölüm 1'i

Eski istekleri silme

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
        ]

Oran sınırlamayı açıklayan akış şemasının Bölüm 2'si

İstek sayısını kontrol et

    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 tarafından yapılan istek sayısı kontrolünü gösteren diyagram: sayı limite eşit veya üstündeyse true, değilse false döner.

Oran sınırı kontrolü ekle

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

Uç noktaya oran sınırı uygula

@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 kez istek gönder:

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"}'

Çıktı:

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

Hadi pratik yapalım!

FastAPI ile Üretime AI Dağıtımı