Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
#!/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 <>.

This comment has been minimized.

Copy link

@iamsuneeth iamsuneeth commented Jun 27, 2018


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 ??


This comment has been minimized.

Copy link

@jonbesga jonbesga commented Aug 12, 2018

This doesn't work.


This comment has been minimized.

Copy link

@jmfrank63 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.


This comment has been minimized.

Copy link
Owner Author

@pquentin 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