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.

Medium Python 3.9+ Aug 9, 2026 System design patterns 13 views 0 copies

Python code

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

stdout
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

  1. Implement a token bucket where tokens are added at a fixed rate instead of draining.
  2. 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

Run this sample

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

Open editor

More from System design patterns

Related tutorials and quizzes for this topic.