You can use threading.Semaphore to count and block requests that exceed the limit, combined with threading.Timer to schedule a function that issues a semaphore.
from threading import Semaphore, Timer from functools import wraps def ratelimit(limit, every): def limitdecorator(fn): semaphore = Semaphore(limit) @wraps(fn) def wrapper(*args, **kwargs): semaphore.acquire() try: return fn(*args, **kwargs) finally:
I expanded this idea and published a library on PyPI called limit .
source share