python - 异步请求退避/节流最佳实践

Question

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

问题 :我现在达到了调用限制，因此在最近 5 个左右的调用中收到 403 个错误。

建议的解决方案 我实现了 try/except在 get_data() 中找到功能。我拨打了电话，然后因为 403: Exceeded call limit 调用不成功我退出 back_off秒并重试至 retries次:

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 , ratelimit和 asyncio-throttle .

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

Answer 1

您可以通过在每个请求之前添加延迟来实现“最多 100 个请求/分钟”。
100 个请求/分钟相当于 1 个请求/0.6 秒。

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()
...