Created
December 7, 2011 01:51
-
-
Save gregburek/1441055 to your computer and use it in GitHub Desktop.
Rate limiting function calls with Python Decorators
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
def RateLimited(maxPerSecond): | |
minInterval = 1.0 / float(maxPerSecond) | |
def decorate(func): | |
lastTimeCalled = [0.0] | |
def rateLimitedFunction(*args,**kargs): | |
elapsed = time.clock() - lastTimeCalled[0] | |
leftToWait = minInterval - elapsed | |
if leftToWait>0: | |
time.sleep(leftToWait) | |
ret = func(*args,**kargs) | |
lastTimeCalled[0] = time.clock() | |
return ret | |
return rateLimitedFunction | |
return decorate | |
@RateLimited(2) # 2 per second at most | |
def PrintNumber(num): | |
print num | |
if __name__ == "__main__": | |
print "This should print 1,2,3... at about 2 per second." | |
for i in range(1,100): | |
PrintNumber(i) |
I added Python 3.12 generic type-hinting and support for recursive calls using an RLock
:
from typing import Callable, cast
from functools import wraps
import threading
import time
def rate_limited[T, **P](max_per_second: int) -> Callable[[Callable[P, T]], Callable[P, T]]:
def decorator(fn: Callable[P, T]) -> Callable[P, T]:
lock = threading.RLock()
min_interval = 1.0 / max_per_second
last_time_called = 0.0
@wraps(fn)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
with lock:
nonlocal last_time_called
elapsed = time.perf_counter() - last_time_called
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
last_time_called = time.perf_counter()
return fn(*args, **kwargs)
return cast(Callable[P, T], wrapper)
return decorator
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
An extended version of @oPromessa 's utility. Thank you for sharing @oPromessa !