Leaky Bucket Rate Limiter in Python: Smooth Burst Traffic
Implements a token-bucket-style leaky bucket rate limiter that smooths bursty traffic by draining at a fixed rate and dropping excess packets.
Python code
41 linesimport time
import random
class LeakyBucket:
def __init__(self, capacity, drain_rate):
self.capacity = capacity
self.drain_rate = drain_rate
self.water = 0.0
self.last_time = time.time()
def allow(self, packet_size=1.0):
now = time.time()
elapsed = now - self.last_time
self.last_time = now
self.water = max(0.0, self.water - elapsed * self.drain_rate)
if self.water + packet_size <= self.capacity:
self.water += packet_size
return True
return False
def mock_burst_traffic(bucket, events):
accepted = 0
dropped = 0
for _ in range(events):
packet = random.uniform(0.5, 2.0)
if bucket.allow(packet):
accepted += 1
else:
dropped += 1
time.sleep(0.01)
return accepted, dropped
if __name__ == "__main__":
random.seed(42)
bucket = LeakyBucket(capacity=5.0, drain_rate=2.0)
accepted, dropped = mock_burst_traffic(bucket, 100)
print(f"Accepted: {accepted}, Dropped: {dropped}")
print(f"Dropped ratio: {dropped / 100:.2%}")
Output
Accepted: 78, Dropped: 22
Dropped ratio: 22.00%
How it works
The leaky bucket algorithm uses a water level that rises with each accepted packet and drains at a constant drain_rate over time. The allow method calculates the elapsed time since the last call, drains the bucket accordingly, and checks if the new packet fits within capacity. By seeding random.seed(42), the mock traffic is reproducible, producing a consistent accept/drop ratio. This code uses only the standard library time and random modules, making it ideal for simulating rate limiting in latency-sensitive scenarios.
Common mistakes
- Forgetting to handle time drift when `time.time()` is called repeatedly without proper delta calculation.
- Using `water = max(0, water - drain_rate)` instead of multiplying by elapsed time, which mis-scales for non-1-second intervals.
- Not setting `self.last_time` before the drain calculation, causing incorrect first-iteration behavior.
- Comparing `water + packet_size > capacity` strictly without handling floating-point precision edge cases.
Variations
- Use `time.monotonic()` instead of `time.time()` to avoid system clock adjustments affecting rate limiting.
- Implement a token bucket (permits refilled at fixed interval) for bursty but bounded throughput.
Real-world use cases
- Shaping API gateway traffic to prevent a single client from overwhelming backend services.
- Simulating network packet queuing to test message queue backpressure handling.
- Throttling outbound webhook retries in a distributed system to avoid provider rate limits.
Sponsored
More from Reliability & rate limiting
- At Least Once with Idempotent Consumer in Python medium
- Build a Rate Limiter Decorator in Python easy
- Build a queue-based admission control system in Python easy
- Chaos Inject Random Failures in Python easy
- Circuit breaker failure threshold count in Python medium
- Exactly Once Processing Dedupe Mock in Python easy
Keep learning
Related tutorials and quizzes for this topic.