Redis-inspired sliding window rate limiter in Python
A pure-Python sliding window rate limiter using a deque of timestamps, mock-ready for Redis-backed production limits.
Python code
30 linesimport time
from collections import deque
class SlidingWindowRateLimiter:
def __init__(self, max_requests: int, window_seconds: int) -> None:
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: dict[str, deque] = {}
def is_allowed(self, client_id: str) -> bool:
now = time.monotonic()
if client_id not in self.requests:
self.requests[client_id] = deque()
window = self.requests[client_id]
while window and now - window[0] >= self.window_seconds:
window.popleft()
if len(window) < self.max_requests:
window.append(now)
return True
return False
if __name__ == "__main__":
limiter = SlidingWindowRateLimiter(max_requests=3, window_seconds=1)
for _ in range(5):
print(f"Allowed: {limiter.is_allowed('user-1')}")
time.sleep(0.1)
time.sleep(1.1)
print(f"After cooldown: {limiter.is_allowed('user-1')}")
Output
Allowed: True
Allowed: True
Allowed: True
Allowed: False
Allowed: False
After cooldown: True
How it works
The collections.deque stores timestamps per client, letting you pop expired entries from the left and append new ones at the right. time.monotonic avoids clock jumps and is meant for measuring intervals, unlike time.time. Checking length against max_requests enforces a strict sliding window instead of a fixed reset period. This mirrors Redis' ZINCRBY + ZREMRANGEBYSCORE approach but uses Python's standard library. For production, swap the internal dict with a Redis-backed store and pipeline commands.
Common mistakes
- Using `time.time()` instead of `time.monotonic()` — system clock changes break the window.
- Not cleaning up deque entries, causing unbounded memory for many client IDs.
- Treating the window as fixed (reset at 0) instead of sliding with each request.
- Forgetting thread-safety — this code is not safe for concurrent access without a lock.
Variations
- Replace the deque with a `sortedcontainers.SortedSet` for very large request volumes and better memory bounds.
- Add a thread lock around `is_allowed` to make it safe for multi-threaded web servers.
Real-world use cases
- Rate limiting per-IP API calls in a Flask/FastAPI gateway to prevent brute-force attacks.
- Throttling third-party API usage per user account to honor vendor quotas.
- Mocking a Redis sliding-window limiter in tests before swapping in the real `zadd`/`zremrangebyscore` implementation.
Sponsored
More from Caching & Redis
- Cache Asides in Python with a Read-Through Loader easy
- Cache Data in Redis with Python easy
- Cache Penetration Null Object Mock in Python medium
- Cache Stampede Prevention with SingleFlight in Python medium
- Cache Warming with Python: Preload Hot Keys easy
- Coalescing duplicate in-flight requests: one shared result for concurrent callers hard
Keep learning
Related tutorials and quizzes for this topic.