Rate limiting

from collections import defaultdict, deque
 
LIMIT, WINDOW = 5, 60                 # 5 requests per 60 seconds per API key
 
# FIXED WINDOW: count per calendar minute. Simple, but allows bursts at the boundary.
fixed = defaultdict(int)
def fixed_window(key: str, t: float) -> bool:
    bucket = (key, int(t // WINDOW))
    fixed[bucket] += 1
    return fixed[bucket] <= LIMIT
 
# SLIDING WINDOW LOG: count requests in the last 60 seconds exactly.
logs = defaultdict(deque)
def sliding_window(key: str, t: float) -> bool:
    q = logs[key]
    while q and q[0] <= t - WINDOW:
        q.popleft()
    if len(q) < LIMIT:
        q.append(t)
        return True
    return False
 
# 5 requests at second 58 and 5 more at second 61: 10 requests within 3 seconds.
times = [58] * 5 + [61] * 5
print("fixed window  :", sum(fixed_window("k1", t) for t in times), "of 10 allowed")
print("sliding window:", sum(sliding_window("k2", t) for t in times), "of 10 allowed")
 
# What a good 429 response tells the client:
def response_for(key: str, t: float) -> tuple[int, dict]:
    ok = sliding_window(key, t)
    remaining = LIMIT - len(logs[key])
    headers = {"RateLimit-Limit": str(LIMIT), "RateLimit-Remaining": str(max(remaining, 0))}
    if not ok:
        headers["Retry-After"] = str(int(logs[key][0] + WINDOW - t))
    return (200 if ok else 429), headers
for t in [0, 1, 2, 3, 4, 5]:
    print(t, *response_for("k3", t))

fixed window  : 10 of 10 allowed
sliding window: 5 of 10 allowed
0 200 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '4'}
1 200 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '3'}
2 200 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '2'}
3 200 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '1'}
4 200 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '0'}
5 429 {'RateLimit-Limit': '5', 'RateLimit-Remaining': '0', 'Retry-After': '55'}