The problem
You need to limit how often a function can be called to respect API rate limits.
The solution
import functools
import time
from collections import defaultdict
from threading import Lock
class RateLimiter:
def __init__(self, calls_per_second: float = 1.0):
self.calls_per_second = calls_per_second
self.min_interval = 1.0 / calls_per_second
self.last_call = defaultdict(float)
self.lock = Lock()
def __call__(self, func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = id(args[0]) if args else 0
with self.lock:
current_time = time.time()
elapsed = current_time - self.last_call[key]
if elapsed < self.min_interval:
sleep_time = self.min_interval - elapsed
time.sleep(sleep_time)
self.last_call[key] = time.time()
return func(*args, **kwargs)
return wrapperParameters
calls_per_secondfloatMaximum calls allowed per second
Put it to work
@RateLimiter(calls_per_second=2)
def call_api(endpoint: str):
print(f"Calling {endpoint} at {time.time()}")
return {"status": "success"}
# Will be rate limited to 2 calls per second
for i in range(5):
call_api(f"/users/{i}")
print(f"Call {i} completed")