what about aiohttp.ClientSession? #71
-
|
Unless I am doing something wrong, it wont work with that. Can you provide an example? Thank you |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 7 replies
-
|
You are gating access to the Gather schedules each entry as a separate task, so they run in parallel independently. You are only rate limiting access to the You can share the limiter between tasks; you could pass it in as an argument: async def fetch(limiter, something):
async with limiter:
await get(something)
await asyncio.gather(*[fetch(limiter, t) for t in range(100)])So when each task runs, each task has to first ask for the rate limiter if it can run the Note that the Bursting section in the documentation uses tasks too; the limiter is used inside the task coroutine. |
Beta Was this translation helpful? Give feedback.
-
|
I have also tried to disable bursting as per docs: |
Beta Was this translation helpful? Give feedback.
-
|
Your output shows it is working just fine:
Taking just the seconds values:
So after the initial burst the limiter let's through one request per 5 seconds == 2 requests per 10 seconds. What makes you think it isn't working? |
Beta Was this translation helpful? Give feedback.
-
This code isn't using the same pattern as your first example. I also can't reproduce your output; your exact code, for me, outputs: >>> l.run_until_complete(fetcher([n for n in range(10)]))
27:32 404
27:37 404
27:42 404
27:47 404
27:52 404
27:57 404
28:02 404
28:07 404
28:12 404
28:17 404Your iPython output is missing a step; where is |
Beta Was this translation helpful? Give feedback.
-
|
I think the library should do what it says on the tin, at any given interval no more than max_requests should be executed, so this burst behaviour to me is not a feature but rather a bug. |
Beta Was this translation helpful? Give feedback.
-
|
The library does what it says on the tin: it implements the leaky bucket algorithm. The documentation explicitly calls out the possibility of bursting and how to handle that. |
Beta Was this translation helpful? Give feedback.
-
|
Right, for now I am resorting to using my own limiter routine, but I still hope that this package gets fixed import asyncio
from datetime import datetime, timedelta
from aiohttp import ClientSession
async def limitCalls(tasks: list, max_rate=1, interval=0):
n = 0
results = []
while _tasks := tasks[n:n+max_rate]:
n += max_rate
t = datetime.now() + timedelta(seconds=interval)
results.extend(await asyncio.gather(*_tasks))
if n < len(tasks):
block = (t - datetime.now()).total_seconds()
block > 0 and await asyncio.sleep(block)
return results
async def fetch(n):
async with ClientSession() as session:
async with session.get(f"https://www.google.co.uk/SORRY_GOOGLE/{n}") as resp:
print(datetime.now().strftime("%H:%M:%S"), resp.status)
return await resp.text()
l = asyncio.get_event_loop()
l.run_until_complete(limitCalls([fetch(n) for n in range(10)], 3, 5))
### EXECUTION
time ./main.py
08:57:30 404
08:57:30 404
08:57:30 404
08:57:35 404
08:57:35 404
08:57:35 404
08:57:40 404
08:57:40 404
08:57:40 404
08:57:45 404
real 0m15.211s |
Beta Was this translation helpful? Give feedback.
-
Sorry, maybe I am missing something, but Please be explicit in how the following script doesn't do what you wanted it to do: import asyncio
from datetime import datetime
from aiolimiter import AsyncLimiter
from aiohttp import ClientSession
async def fetch(session, limiter, n):
async with limiter:
async with session.get(f"https://www.google.co.uk/SORRY_GOOGLE/{n}") as resp:
print(datetime.now().strftime("%H:%M:%S"), resp.status)
return await resp.text()
async def main():
limiter = AsyncLimiter(3, 5)
async with ClientSession() as session:
await asyncio.gather(*(fetch(session, limiter, n) for n in range(10)))
asyncio.run(main())Here the limiter allows for 3 requests across a 5 second window: Your script groups 3 requests every 5 seconds, and delays creating tasks. With the limiter, the tasks are all created up front but then the If you wanted to see a burst of 3 requests every 5 seconds, then rate-limit creating 3 tasks at a time: import asyncio
from datetime import datetime
from aiolimiter import AsyncLimiter
from aiohttp import ClientSession
async def fetch(session, n):
async with session.get(f"https://www.google.co.uk/SORRY_GOOGLE/{n}") as resp:
print(datetime.now().strftime("%H:%M:%S"), resp.status)
return await resp.text()
async def main():
limiter = AsyncLimiter(1, 5)
async with ClientSession() as session:
tasks = [fetch(session, n) for n in range(10)]
while tasks:
async with limiter:
to_run, tasks = tasks[:3], tasks[3:]
await asyncio.gather(*to_run)
asyncio.run(main())which outputs: |
Beta Was this translation helpful? Give feedback.
You are gating access to the
asyncio.gather()call. You haven't made it explicit, but I'm going to assume that that's not what you wanted, but that you instead wanted to limitget()calls.Gather schedules each entry as a separate task, so they run in parallel independently. You are only rate limiting access to the
asyncio.gather()call, and each time execution has reached that line, 100 tasks are created. Those tasks are themselves not rate limited. If you want to rate limit theget()calls, put yourwithblock there, inside the task function.You can share the limiter between tasks; you could pass it in as an argument: