Skip to content

Instantly share code, notes, and snippets.

@jsbueno
Last active August 19, 2021 21:32
Show Gist options
  • Save jsbueno/43314724c02d05acfef7079b9c4dc152 to your computer and use it in GitHub Desktop.
Save jsbueno/43314724c02d05acfef7079b9c4dc152 to your computer and use it in GitHub Desktop.
Rate Limiter for async APIs. Juste create one guard for different API in your process. Not multiprocess safe!
def create_rate_limit_guard(rate_limit=1200, safety_margin=0.9):
"""Rate limit is given in maximum requests per minute.
"""
# TBD: it would easy to have code to throttle by maximum active requests
# instead of total requests per minute.
# I will just let the accounting of concurrent_requests in place, though
class Guard:
request_interval = (60 / rate_limit) * safety_margin
current_requests = 0
max_concurrent_requests = 0
last_request = 0
async def __aenter__(self):
cls = self.__class__
cls.current_requests += 1
if (throttle_wait:= time.time() - last_request) < cls.request_interval:
await asyncio.sleep(throttle_wait)
cls.current_requests += 1
cls.last_request = time.time()
async def __aexit__(self, exc_type, exc, tb):
cls = self.__class__
cls.max_concurrent_requests = max(cls.max_concurrent_requests, cls.current_requests)
cls.current_requests -= 1
return Guard
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment