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.

Medium Python 3.8+ Aug 9, 2026 Caching & Redis 14 views 0 copies

Python code

37 lines
Python 3.8+
import 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

stdout
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

  1. Use a Redis Lua script with INCRBY and EXPIRE to atomically handle the counter and leak in production
  2. 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

Run this sample

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

Open editor

More from Caching & Redis

Related tutorials and quizzes for this topic.