Last active
August 19, 2021 21:32
-
-
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!
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
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