Skip to content

Instantly share code, notes, and snippets.

@niro1987
Created January 15, 2022 10:30
Show Gist options
  • Save niro1987/fc1acb1b2a16c5458a06be1d275fd7ed to your computer and use it in GitHub Desktop.
Save niro1987/fc1acb1b2a16c5458a06be1d275fd7ed to your computer and use it in GitHub Desktop.
Python asyncio RateLimiter
import asyncio
class RateLimiter:
def __init__(self, max_calls: int = 100, interval: int = 60):
self.max_calls = max_calls
self.interval = interval
self.queue = asyncio.Queue(max_calls)
self.worker = asyncio.run_coroutine_threadsafe(self.queue_handler())
async def __aenter__(self):
await self.queue.put(self.interval)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
return None
async def sleep(self):
await asyncio.sleep(self.interval)
await self.queue.get()
self.queue.task_done()
async def queue_handler(self):
while True:
tasks = [self.sleep() for _ in range(self.queue.qsize())]
await asyncio.gather(*tasks)
async def main():
limiter = RateLimiter(2, 1)
for i in range(10):
async with limiter:
print("Hi", i)
if __name__ == "__main__":
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment