Giới hạn tốc độ (Rate Limiting)

Triển khai AI vào môi trường sản xuất với FastAPI

Matt Eckerle

Software and Data Engineering Leader

Giới thiệu về rate limiting

một đèn giao thông

Mục đích: Kiểm soát tần suất yêu cầu API.
Phản hồi: Trả về HTTP 429 ("Too Many Requests") khi vượt giới hạn.

Cách rate limiting hoạt động

Phần 1 của sơ đồ luồng giải thích rate limiting

Xác thực thông tin đăng nhập đến

Phần 2 của sơ đồ luồng giải thích rate limiting

Kiểm tra rate limiting

Phần 3 của sơ đồ luồng giải thích rate limiting

Thiết lập API của bạn

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"

Logic của bộ giới hạn tốc độ

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

Phần 1 của sơ đồ luồng về logic triển khai rate limiting

Xóa yêu cầu cũ

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
        ]

Phần 2 của sơ đồ luồng giải thích rate limiting

Kiểm tra số yêu cầu

    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

Sơ đồ kiểm tra số lượng yêu cầu do API gửi: nếu đếm vượt hoặc bằng giới hạn, trả về true; ngược lại trả về false.

Thêm kiểm tra giới hạn tốc độ

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

Áp dụng rate limit cho 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
    }

Gửi yêu cầu 11 lần:

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

Kết quả:

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

Cùng luyện tập nào!

Triển khai AI vào môi trường sản xuất với FastAPI