Rate Limiting: A Practical Guide from Scratch

API crashes usually happen because one rogue client or a sudden traffic spike slams the backend, so implementing rate limiting is non-negotiable for any stable deployment. Depending on your traffic patterns, some algorithms handle bursts better than others.

Strategy 1: Fixed Window Counter

This is the most basic approach. You track requests within a rigid time block (e.g., 60 seconds). If the counter hits the limit, further requests are blocked until the next window starts.

import time
from collections import defaultdict

class FixedWindowLimiter:
 def __init__(self, max_requests=10, window_seconds=60):
 self.max_requests = max_requests
 self.window_seconds = window_seconds
 self.requests = defaultdict(list)

 def allow_request(self, user_id):
 now = time.time()
 window_start = now - self.window_seconds
 # Remove old entries
 self.requests[user_id] = [t for t in self.requests[user_id] if t > window_start]
 if len(self.requests[user_id]) >= self.max_requests:
 return False
 self.requests[user_id].append(now)
 return True

The main issue here is the "boundary problem." A user could exhaust their limit at the very end of window A and immediately use another full quota at the start of window B, effectively doubling their throughput in a short burst.

Strategy 2: Sliding Window Log

To fix the boundary spike, the sliding window log keeps a precise timestamp of every request. It only counts requests that happened within the last X seconds from the current millisecond.

import time
from collections import defaultdict

class SlidingWindowLimiter:
 def __init__(self, max_requests=10, window_seconds=60):
 self.max_requests = max_requests
 self.window_seconds = window_seconds
 self.logs = defaultdict(list)

 def allow_request(self, user_id):
 now = time.time()
 cutoff = now - self.window_seconds
 # Remove timestamps outside the window
 self.logs[user_id] = [t for t in self.logs[user_id] if t > cutoff]
 if len(self.logs[user_id]) >= self.max_requests:
 return False
 self.logs[user_id].append(now)
 return True

This is much smoother, but memory is the trade-off. Storing every single timestamp for millions of users will eat up your RAM quickly.

Strategy 3: Token Bucket

This is my go-to for real-world AI workflow integrations. Imagine a bucket that refills with tokens at a constant rate. Requests consume tokens. If the bucket is empty, the request is rejected.

import time

class TokenBucket:
 def __init__(self, capacity=10, refill_rate=1):
 self.capacity = capacity
 self.refill_rate = refill_rate # tokens per second
 self.tokens = capacity
 self.last_refill = time.time()

 def allow_request(self):
 now = time.time()
 elapsed = now - self.last_refill
 self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
 self.last_refill = now
 if self.tokens >= 1:
 self.tokens -= 1
 return True
 return False

It's memory-efficient and handles bursts beautifully while maintaining a strict long-term average.

Implementation Tips

• Distributed State: In-memory lists only work for single-instance apps. For any real production deployment, move your counters to Redis.
• Client Communication: Don't just return a 429 error. Always include X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset headers so the client knows when to back off.
• Cleanup: If you use in-memory maps, you must implement a TTL or a cleanup job, otherwise, your memory usage will grow linearly with your user base.