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.

Medium Python 3.9+ Aug 9, 2026 Reliability & rate limiting 14 views 0 copies

Python code

41 lines
Python 3.9+
import 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

stdout
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

  1. Use `time.monotonic()` instead of `time.time()` to avoid system clock adjustments affecting rate limiting.
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Reliability & rate limiting

Related tutorials and quizzes for this topic.