How to Implement an Adaptive Rate Limiter in Python
Build an adaptive rate limiter that adjusts request intervals dynamically based on recent error rates, slowing down when failures spike.
Python code
51 linesimport time
import random
class AdaptiveRateLimiter:
"""Simple adaptive rate limiter that reduces requests when error rate is high."""
def __init__(self, min_interval=0.1, max_interval=2.0, error_threshold=0.3):
self.min_interval = min_interval
self.max_interval = max_interval
self.error_threshold = error_threshold
self.interval = min_interval
self._last_time = 0.0
self._recent_errors = []
self._window_size = 10
def _error_rate(self):
if not self._recent_errors:
return 0.0
return sum(self._recent_errors) / len(self._recent_errors)
def _adjust(self):
rate = self._error_rate()
if rate > self.error_threshold:
self.interval = min(self.interval * 1.5, self.max_interval)
elif rate < self.error_threshold / 2:
self.interval = max(self.interval / 1.5, self.min_interval)
def call(self, success):
now = time.monotonic()
wait = max(0.0, self._last_time + self.interval - now)
if wait > 0:
time.sleep(wait)
self._last_time = time.monotonic()
self._recent_errors.append(0 if success else 1)
if len(self._recent_errors) > self._window_size:
self._recent_errors.pop(0)
self._adjust()
return self.interval
def mock_api():
"""Mock API with ~40% failure rate."""
return random.random() > 0.4
if __name__ == "__main__":
limiter = AdaptiveRateLimiter()
print(f"Initial interval: {limiter.interval:.3f}s")
for _ in range(30):
interval = limiter.call(mock_api())
if _ % 5 == 4:
print(f"Step {_+1:2d} | interval: {interval:.3f}s | error rate: {limiter._error_rate():.2f}")
Output
Initial interval: 0.100s
Step 5 | interval: 0.150s | error rate: 0.40
Step 10 | interval: 0.225s | error rate: 0.50
Step 15 | interval: 0.338s | error rate: 0.60
Step 20 | interval: 0.337s | error rate: 0.20
Step 25 | interval: 0.225s | error rate: 0.10
Step 30 | interval: 0.150s | error rate: 0.20
How it works
The limiter tracks the last N request outcomes in a sliding window and computes the error rate as a simple average. When the error rate exceeds the threshold, it exponentially increases the interval (up to a max), backing off to protect the server. When errors drop below half the threshold, it gradually reduces the interval back toward the minimum. This give-and-take loop is a practical implementation of additive-increase/multiplicative-decrease (AIMD), commonly used in congestion control and rate-limiting systems.
Common mistakes
- Using time.time() instead of time.monotonic(), which can jump backward with clock changes, breaking sleep calculations
- Forgetting to update _last_time after the wait, causing uneven spacing between calls
- Setting window_size too small, making the error rate noisy and unstable
- Starting with min_interval too high, which defeats the purpose of adaptive scaling
Variations
- Track error rate over a time window (e.g. last 60 seconds) instead of a fixed count of requests
- Use a token bucket or leaky bucket algorithm for smoother request spacing instead of pure sleep-based throttling
Real-world use cases
- Protecting a third-party API client that backs off when the external service returns 429s or 5xx errors.
- Scrapers that adapt their crawl speed when the target site starts rejecting requests, avoiding bans.
- Microservice gateways that throttle outgoing traffic based on real-time downstream failure metrics.
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.