Skip to content

Instantly share code, notes, and snippets.

@hugokernel
Last active December 2, 2021 19:24
Show Gist options
  • Save hugokernel/c478d639849e1e772b1395a546100031 to your computer and use it in GitHub Desktop.
Save hugokernel/c478d639849e1e772b1395a546100031 to your computer and use it in GitHub Desktop.
Async rate limit call
def ratelimit(seconds=0, minutes=None, hours=None, days=None):
"""Rate limit call to a function"""
for var, delay in (
('minutes', 60),
('hours', 60 * 60),
('days', 60 * 60 * 24),
):
if locals()[var] is not None:
seconds += locals()[var] * delay
timeout = 0
def decorator(func):
async def wrapper(*args, **kwargs):
nonlocal timeout
if time.time() < timeout:
return
await func(*args, **kwargs)
# No call before `timeout`
timeout = time.time() + seconds
return wrapper
return decorator
# Example:
import asyncio
import time
@ratelimit(seconds=2)
async def foo(name):
await asyncio.sleep(1)
print(name)
async def main():
await foo("OH!")
await foo("Skipped")
time.sleep(2)
await foo("OH!!")
await foo("Skipped")
time.sleep(3)
await foo("OH!!!")
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
# Output:
# OH!
# OH!!
# OH!!!
@yeus
Copy link

yeus commented Dec 2, 2021

I created a similar asyncio version but using the token bucket algorithm for
max number of calls during/time interval which results in a maximum burst rate. it can be combined with the above rate limiter without burst rate..

https://gist.github.com/yeus/dff02dce88c6da9073425b5309f524dd

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment