Pythonrate_limiter.py
Copy
1
import time
2
from collections import deque
3
from fastapi import FastAPI, Request
4
from fastapi.responses import JSONResponse
5
 
6
app = FastAPI()
7
 
8
class SlidingWindowRateLimiter:
9
    def __init__(self, max_requests: int, window_seconds: int):
10
        self.max_requests = max_requests
11
        self.window_seconds = window_seconds
12
        self.clients: dict[str, deque[float]] = {}
13
 
14
    def is_allowed(self, client_id: str) -> tuple[bool, int]:
15
        now = time.monotonic()
16
        window_start = now - self.window_seconds
17
 
18
        if client_id not in self.clients:
19
            self.clients[client_id] = deque()
20
 
21
        timestamps = self.clients[client_id]
22
        # Remove old timestamps
23
        while timestamps and timestamps[0] < window_start:
24
            timestamps.popleft()
25
 
26
        if len(timestamps) >= self.max_requests:
27
            retry_after = int(timestamps[0] - window_start) + 1
28
            return False, retry_after
29
 
30
        timestamps.append(now)
31
        remaining = self.max_requests - len(timestamps)
32
        return True, remaining
33
 
34
limiter = SlidingWindowRateLimiter(max_requests=10, window_seconds=60)
35
 
36
@app.middleware("http")
37
async def rate_limit_middleware(request: Request, call_next):
38
    client_ip = request.client.host or "unknown"
39
    allowed, value = limiter.is_allowed(client_ip)
40
 
41
    if not allowed:
42
        return JSONResponse(
43
            status_code=429,
44
            content={"error": "Rate limit exceeded"},
45
            headers={
46
                "X-RateLimit-Limit": str(limiter.max_requests),
47
                "X-RateLimit-Remaining": "0",
48
                "Retry-After": str(value),
49
            },
50
        )
51
 
52
    response = await call_next(request)
53
    response.headers["X-RateLimit-Limit"] = str(limiter.max_requests)
54
    response.headers["X-RateLimit-Remaining"] = str(value)
55
    return response
56
 
57
@app.get("/data")
58
async def get_data():
59
    return {"data": "some response"}
60