Designing a Rate Limiter in Python

# Token bucket rate limiter using Redis
import time
import redis.asyncio as aioredis
from fastapi import Request, HTTPException

redis = aioredis.from_url("redis://localhost")

async def check_rate_limit(api_key: str, limit: int = 100, window: int = 60):
    """Sliding window rate limiter using Redis sorted sets."""
    key = f"rate:{api_key}"
    now = time.time()
    window_start = now - window

    pipe = redis.pipeline()
    # Remove old requests outside the window
    pipe.zremrangebyscore(key, 0, window_start)
    # Count requests in window
    pipe.zcard(key)
    # Add current request
    pipe.zadd(key, {str(now): now})
    # Set expiry
    pipe.expire(key, window)

    _, count, _, _ = await pipe.execute()

    if count >= limit:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={"Retry-After": str(window), "X-RateLimit-Limit": str(limit)}
        )

    return {"remaining": limit - count - 1}

# FastAPI middleware
from fastapi import FastAPI
from fastapi.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        api_key = request.headers.get("X-API-Key", request.client.host)
        await check_rate_limit(api_key)
        return await call_next(request)

app = FastAPI()
app.add_middleware(RateLimitMiddleware)