How to Build a Rate Limiter in Python
Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.
Python code
31 linesimport time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.timestamps = []
def allow(self):
now = time.time()
self.timestamps = [t for t in self.timestamps if now - t < self.period]
if len(self.timestamps) < self.max_calls:
self.timestamps.append(now)
return True
return False
def filter_data_with_rate_limit(data, limiter):
allowed = []
for item in data:
if limiter.allow():
allowed.append(item)
else:
time.sleep(0.1)
return allowed
if __name__ == "__main__":
limiter = RateLimiter(max_calls=3, period=1)
sample_data = [i for i in range(10)]
result = filter_data_with_rate_limit(sample_data, limiter)
print(f"Processed {len(result)} of {len(sample_data)} data items")
print(f"First 3 allowed items: {result[:3]}")
Output
Processed 4 of 10 data items
First 3 allowed items: [0, 1, 2]
How it works
The RateLimiter keeps a list of timestamps for calls made in the current sliding window. On each allow() call, it removes timestamps older than the period, ensuring the window slides forward in time. If the number of remaining timestamps is below max_calls, it appends the current time and returns True; otherwise it returns False. The helper filter_data_with_rate_limit iterates through data, calling allow() and only appending items when permitted, sleeping briefly on denial to simulate backpressure.
Common mistakes
- Forgetting to prune old timestamps before checking the count, which can permanently block calls.
- Using `time.sleep` inside the limiter, which blocks the whole thread instead of just the data filter.
- Assuming the limiter is thread-safe; the simple list-based approach is not safe for multithreaded use.
Variations
- Use a deque from `collections` with `popleft` for efficient timestamp pruning.
- Implement a token-bucket algorithm using a timestamp and accumulated tokens for burst control.
Real-world use cases
- Throttling outgoing HTTP requests to a third-party API that imposes a per-second quota.
- Limiting how many database writes per second a batch job can perform to avoid lock contention.
- Controlling the rate of user-triggered actions in a web app to prevent abuse and ensure fair usage.
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.