Token Bucket Rate Limiter

0.3.0 · active · verified Sun Apr 12

The `token-bucket` package provides a very fast implementation of the token bucket algorithm, suitable for rate limiting in web applications. It manages bucket state internally without requiring a separate timer thread, allowing for controlled bursts while maintaining a consistent average rate. The current version is 0.3.0, and releases appear to be infrequent, focusing on stability and core algorithm improvements.

Warnings

breaking Version 0.3.0 dropped support for Python 2.x and now requires Python 3.5 or newer. Users on older Python versions must upgrade their environment or stick to `token-bucket<0.3.0`.
Fix: Ensure your project runs on Python 3.5+; if not, upgrade Python or pin `token-bucket` to version 0.2.x.
gotcha The `token-bucket` library primarily provides an in-memory, single-process token bucket implementation. For distributed rate limiting across multiple application instances (e.g., in a microservices architecture), you will need to implement a shared state mechanism (e.g., using Redis) or use a different library designed for distributed rate limiting.
Fix: For distributed systems, consider integrating with an external store like Redis or exploring libraries like `token-throttler` or `bucketflow` that offer distributed backend support.
gotcha Prior to version 0.3.0, rates less than 1.0 token per second might not have been handled correctly. Version 0.3.0 explicitly added support for fractional rates, allowing for finer-grained control over low-frequency limits.
Fix: Upgrade to version 0.3.0 or later to ensure accurate rate limiting when `rate` is less than 1.0 token per second.

Install

pip install token-bucket Install stable version

Imports

TokenBucket
```
from token_bucket import TokenBucket
```
The main class for creating and managing a token bucket.

Quickstart

Initialize a `TokenBucket` with a `capacity` and `rate` (tokens per second). Use the `consume()` method to attempt to acquire tokens. It returns `True` if tokens are successfully acquired, `False` otherwise. The bucket automatically replenishes tokens based on the elapsed time since the last operation, up to its capacity.

from token_bucket import TokenBucket
import time

# Create a token bucket with a capacity of 10 tokens,
# refilling at a rate of 2 tokens per second.
# This allows for a burst of 10 requests, then 2 requests/second.
bucket = TokenBucket(capacity=10, rate=2.0)

print(f"Initial tokens: {bucket.tokens}")

# Try to consume 5 tokens
if bucket.consume(5):
    print("Consumed 5 tokens successfully.")
else:
    print("Failed to consume 5 tokens.")

print(f"Tokens after first consumption: {bucket.tokens}")

# Wait a bit for tokens to refill
time.sleep(2.0)

# Try to consume 8 tokens (after 2 seconds, 4 tokens should have refilled)
if bucket.consume(8):
    print("Consumed 8 tokens successfully after refill.")
else:
    print("Failed to consume 8 tokens after refill.")

print(f"Tokens after second consumption: {bucket.tokens}")

view raw JSON →