异步请求回退/限制最佳实践

Question

场景：我需要从web应用程序的API收集分页数据，该API的调用限制为每分钟100次。我需要返回的API对象每页包含100个项目，总共105个页面，并且在不断增长（~10500个项目）。同步代码检索所有页面大约需要15分钟，因此不必担心达到调用限制。但是，我想加快数据检索速度，因此我使用asyncio和aiohttp实现了异步调用。数据现在可以在15秒内下载-很好。

问题：我现在达到呼叫限制，因此在过去的5个左右的呼叫中收到403个错误。

建议的解决方案我实现了get_data（）函数中的try/except。我拨打电话，然后当由于403:超出呼叫限制而导致呼叫不成功时，我会后退秒，然后重试最多次：

async def get_data(session, url):
    retries = 3
    back_off = 60  # seconds to try again
    for _ in range(retries):
        try:
            async with session.get(url, headers=headers) as response:
                if response.status != 200:
                    response.raise_for_status()
                print(retries, response.status, url)
                return await response.json()
        except aiohttp.client_exceptions.ClientResponseError as e:
            retries -= 1
            await asyncio.sleep(back_off)
            continue

async def main():
    async with aiohttp.ClientSession() as session:
        attendee_urls = get_urls('attendee') # returns list of URLs to call asynchronously in get_data()
        attendee_data = await asyncio.gather(*[get_data(session, attendee_url) for attendee_url in attendee_urls])
        return attendee_data

if __name__ == '__main__':
    data = asyncio.run(main())

问题：我如何限制aiohttp呼叫，使其保持在每分钟100个呼叫的阈值之下，而无需发出403请求以退出？我尝试了以下模块，但没有一个模块能起到任何作用：ratelimiter、ratelimiter和asyncio throttle。

目标：每分钟进行100次异步调用，但在必要时后退并重试（403：超出调用限制）。




Answer 1

通过在每个请求之前添加延迟，可以实现“最多100个请求/分钟”。

100次请求/分钟相当于1次请求/0.6s。

async def main():

    async with aiohttp.ClientSession() as session:
        attendee_urls = get_urls('attendee') # returns list of URLs to call asynchronously in get_data()
        coroutines = []
        for attendee_url in attendee_urls:
            coroutines.append(get_data(session, attendee_url))
            await asyncio.sleep(0.6)
        attendee_data = asyncio.gather(*coroutines)
        return attendee_data

除了请求速率限制外，API通常还限制同时请求的数量。如果是这样，可以使用BoundedSempahore。

async def main():
    sema = asyncio.BoundedSemaphore(50) # Assuming a concurrent requests limit of 50
...
            coroutines.append(get_data(sema, session, attendee_url))
...

def get_data(sema, session, attendee_url):

...

    for _ in range(retries):
        try:
            async with sema:
                response = await session.get(url, headers=headers):
                if response.status != 200:
                    response.raise_for_status()
...