Skip to content

Instantly share code, notes, and snippets.

Last active March 16, 2024 16:53
Show Gist options
  • Star 32 You must be signed in to star a gist
  • Fork 7 You must be signed in to fork a gist
  • Save pquentin/5d8f5408cdad73e589d85ba509091741 to your computer and use it in GitHub Desktop.
Save pquentin/5d8f5408cdad73e589d85ba509091741 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
import asyncio
import time
import aiohttp
START = time.monotonic()
class RateLimiter:
"""Rate limits an HTTP client that would make get() and post() calls.
Calls are rate-limited by host.
This class is not thread-safe."""
RATE = 1 # one request per second
def __init__(self, client):
self.client = client
self.tokens = self.MAX_TOKENS
self.updated_at = time.monotonic()
async def get(self, *args, **kwargs):
await self.wait_for_token()
now = time.monotonic() - START
print(f'{now:.0f}s: ask {args[0]}')
return self.client.get(*args, **kwargs)
async def wait_for_token(self):
while self.tokens < 1:
await asyncio.sleep(0.1)
self.tokens -= 1
def add_new_tokens(self):
now = time.monotonic()
time_since_update = now - self.updated_at
new_tokens = time_since_update * self.RATE
if self.tokens + new_tokens >= 1:
self.tokens = min(self.tokens + new_tokens, self.MAX_TOKENS)
self.updated_at = now
async def fetch_one(client, i):
url = f'{i}'
# Watch out for the extra 'await' here!
async with await client.get(url) as resp:
resp = await resp.json()
now = time.monotonic() - START
print(f"{now:.0f}s: got {resp['args']}")
async def main():
async with aiohttp.ClientSession() as client:
client = RateLimiter(client)
tasks = [asyncio.ensure_future(fetch_one(client, i)) for i in range(20)]
await asyncio.gather(*tasks)
if __name__ == '__main__':
# Requires Python 3.7+
# This work is licensed under the terms of the MIT license.
# For a copy, see <>.
Copy link


I have a doubt in the execution of this. If I have an array of API calls, how will I use this code to limit them? If I use the same client for all the requests, all the requests will be waiting on same tokens variable and will be executed when the self.tokens is updated in any of the add_new_tokens calls. Am I missing something ??

Copy link

This doesn't work.

Copy link

jmfrank63 commented Aug 16, 2018

When replacing asyncio with trio and integrating this into an existing class this works like a charm. Very efficient algorithm implementation.

Copy link

pquentin commented Sep 6, 2018

Sorry, I don't receive notifications for gists (that's a limitation of gists): prefer commenting on the blog post so that I can answer.

@iamsuneeth Only one request can be sent at the same time, the others will be in asyncio.sleep(). Does that make sense?

@jonbesga I added real parallel execution to the example (and updated it to use Python 3.7). This should convince you that this actually works. (I've been using this in production for months.)

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