Redis Leaky Bucket Rate Limiting Mock in Python
Simulates a Redis-backed leaky bucket rate limiter using a local class with continuous leaking and token capacity checks.
Python code
37 linesimport time
from collections import deque
class LeakyBucket:
def __init__(self, capacity, leak_rate):
self.capacity = capacity
self.leak_rate = leak_rate
self.water = 0.0
self.timestamp = time.time()
self.history = deque()
def allow(self):
current = time.time()
elapsed = current - self.timestamp
self.water = max(0, self.water - elapsed * self.leak_rate)
self.timestamp = current
if self.water + 1 <= self.capacity:
self.water += 1
self.history.append(current)
return True
return False
def water_level(self):
current = time.time()
elapsed = current - self.timestamp
return max(0, self.water - elapsed * self.leak_rate)
if __name__ == "__main__":
bucket = LeakyBucket(capacity=3, leak_rate=1)
for _ in range(5):
print(bucket.allow())
time.sleep(2)
print("Water level after 2s:", bucket.water_level())
print(bucket.allow())
Output
True
True
True
False
False
Water level after 2s: 0.0
True
How it works
The leaky bucket algorithm allows requests at a steady rate, draining water over time based on elapsed seconds multiplied by the leak rate. Each allow() call recalculates the current water level from the elapsed time and timestamp, then checks if capacity remains for another request. The history deque could store timestamps for external logging or inspection, but it isn't used for the decision logic here. This local implementation mimics Redis Lua scripts that atomically update a key's counter and TTL for rate limiting, providing a testable mock for applications.
Common mistakes
- Forgetting to subtract leaked water on every check, leading to immediate false negatives after a burst
- Using integer arithmetic instead of floats, truncating leak amounts and skewing allowed bursts
- Resetting the timestamp after a leak calculation, causing double-counted elapsed time
Variations
- Use a Redis Lua script with INCRBY and EXPIRE to atomically handle the counter and leak in production
- Implement a token bucket instead, filling tokens at a fixed rate rather than draining water
Real-world use cases
- Protecting an API endpoint from traffic spikes by capping request rates per user or IP via a shared store like Redis.
- Controlling concurrency in a message consumer, ensuring a steady ingestion rate without overwhelming downstream services.
- Implementing fair usage policies for SaaS features, like limiting file uploads or exports per account.
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.