Simulate a Leaky Bucket Rate Limiter in Python
This code implements a leaky bucket rate limiter that drains at a fixed rate and accepts or rejects incoming requests based on capacity.
Python code
33 linesimport time
from collections import deque
class LeakyBucket:
"""Simulates a leaky bucket rate limiter with a fixed drain rate."""
def __init__(self, capacity, drain_rate_per_sec):
self.capacity = capacity
self.drain_rate = drain_rate_per_sec
self.water = 0.0
self.last_refill = time.monotonic()
def add_request(self, amount=1):
"""Return True if request fits, False if bucket overflows."""
now = time.monotonic()
elapsed = now - self.last_refill
self.water = max(0.0, self.water - elapsed * self.drain_rate)
self.last_refill = now
if self.water + amount <= self.capacity:
self.water += amount
return True
return False
if __name__ == "__main__":
bucket = LeakyBucket(capacity=3, drain_rate_per_sec=1)
for i in range(6):
accepted = bucket.add_request()
print(f"Request {i}: {'accepted' if accepted else 'rejected'} (water={bucket.water:.2f})")
time.sleep(1.5)
accepted = bucket.add_request()
print(f"Request after drain: {'accepted' if accepted else 'rejected'} (water={bucket.water:.2f})")
Output
Request 0: accepted (water=1.00)
Request 1: accepted (water=2.00)
Request 2: accepted (water=3.00)
Request 3: rejected (water=3.00)
Request 4: rejected (water=3.00)
Request 5: rejected (water=3.00)
Request after drain: accepted (water=1.50)
How it works
The LeakyBucket class tracks a water level that represents the current bucket fill. On each add_request call, it computes the elapsed time since the last refill and drains the bucket by elapsed * drain_rate using time.monotonic() for precise timing. If the remaining water plus the new request fits within capacity, the request is accepted and water is incremented; otherwise it's rejected. The monotonic clock avoids jumps from system time changes, making the simulation reliable for testing rate-limit logic.
Common mistakes
- Using `time.time()` instead of `time.monotonic()` which can cause inaccurate draining during clock adjustments.
- Forgetting to update `last_refill` even when a request is rejected, leading to incorrect drainage on subsequent calls.
- Allowing water to go negative by not clamping to zero with `max()` protocol.
Variations
- Implement a token bucket where tokens are added at a fixed rate instead of draining.
- Use a `deque` to track timestamps of accepted requests and reject if the queue length exceeds capacity.
Real-world use cases
- Rate limiting API endpoints to prevent abuse by capping requests per second.
- Throttling outbound requests to third-party services to respect their quotas.
- Controlling data flow in message queues or streams to avoid overwhelming downstream consumers.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.