How to Build a Rate Limiter in Python
A beginner-friendly token bucket rate limiter with retry logic for handling API rate limits in Python.
Python code
54 linesimport time
import random
class RateLimiter:
"""Simple token bucket rate limiter for beginners."""
def __init__(self, max_tokens=5, refill_rate=1.0):
self.max_tokens = max_tokens
self.tokens = max_tokens
self.refill_rate = refill_rate # tokens per second
self.last_refill = time.monotonic()
def _refill(self):
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(self.max_tokens, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
def acquire(self):
"""Try to take a token. Return True if allowed, False if rate-limited."""
self._refill()
if self.tokens >= 1:
self.tokens -= 1
return True
return False
def fetch_data_with_retry(data_loader, retries=3, max_delay=2.0):
"""Fetch data with retry logic for failed attempts."""
for attempt in range(retries):
try:
return data_loader()
except Exception as e:
wait = random.uniform(0.5, max_delay) * (attempt + 1)
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait:.1f}s...")
time.sleep(wait)
raise RuntimeError("All retry attempts exhausted")
if __name__ == "__main__":
# Example usage
limiter = RateLimiter(max_tokens=3, refill_rate=1)
for i in range(5):
if limiter.acquire():
print(f"Request {i + 1}: allowed")
else:
print(f"Request {i + 1}: rate-limited")
print("\nWaiting 2 seconds for refill...")
time.sleep(2)
if limiter.acquire():
print("Request after refill: allowed")
Output
Request 1: allowed
Request 2: allowed
Request 3: allowed
Request 4: rate-limited
Request 5: rate-limited
Waiting 2 seconds for refill...
Request after refill: allowed
How it works
The RateLimiter class uses a token bucket algorithm where tokens accumulate at a fixed rate over time. Each acquire() call first refills tokens based on elapsed time, then checks if at least one token is available. The _refill method calculates how many tokens to add based on elapsed * refill_rate, capped at max_tokens. The retry function uses exponential backoff with random jitter to space out retry attempts, making it more resilient against temporary failures. This pattern is the foundation for production rate limiters that protect APIs from being overwhelmed.
Common mistakes
- Forgetting to call `_refill()` before checking token availability, causing starvation
- Using `time.time()` instead of `time.monotonic()` which can jump with system clock changes
- Not capping tokens at `max_tokens`, allowing buckets to grow unboundedly
- Setting `retries` too high without exponential backoff, hammering the API
Variations
- Use `threading.Lock` to make `acquire()` thread-safe for concurrent access
- Replace token bucket with fixed window or sliding window algorithms for simpler use cases
Real-world use cases
- Pacing API calls to third-party services like Twitter or GitHub that enforce per-minute limits.
- Protecting internal microservices from request spikes by adding rate limiting at the load balancer layer.
- Implementing client-side rate limiting for bulk data exports to avoid server 429 responses.
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.