1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
# ruff: noqa: T201, G004
"""
Example of using pyrate_limiter with aiohttp.
"""
import asyncio
import logging
import time
from pyrate_limiter import Duration, limiter_factory
from pyrate_limiter.extras.aiohttp_limiter import RateLimitedSession
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
logger.setLevel(logging.DEBUG)
async def test_aiohttp():
url = "https://httpbin.org/get"
limiter = limiter_factory.create_inmemory_limiter(
rate_per_duration=3,
duration=Duration.SECOND,
)
async with RateLimitedSession(limiter) as session:
async def wrapped(i):
logger.info(f"Request {i} at {round(time.time(), 2)}")
resp = await session.get(url)
logger.info(f"Response {i} at {round(time.time(), 2)}: {resp.status}")
return resp
await asyncio.gather(*(wrapped(i) for i in range(10)))
if __name__ == "__main__":
asyncio.run(test_aiohttp())
|
