如何使用异步处理速率限制？

Question

我试图通过API发送请求，API有5个请求/秒。我有以下脚本：

for object_id, object_coors in object_data.items():
           
            time.sleep(1)

            try:
                # fire off the POST request
                search_result = \
                    requests.post(
                        'https://api.planet.com/data/v1/quick-search', auth=HTTPBasicAuth(API_KEY, ''),

                        json=search_request, verify=False)

            except Exception as e:
                print(e)
                time.sleep(5)
                print("Trying again to fire of the POST request")
                search_result = \
                    requests.post(
                        'https://api.planet.com/data/v1/quick-search',
                        auth=HTTPBasicAuth(API_KEY, ''),
                        json=search_request, verify=False)
           task_visual = asyncio.ensure_future(fire_post(session, payload_visual)) 
           tasks_visual.append(task_visual)
           time.sleep(2)
           task_analytical = asyncio.ensure_future(
                        fire_post(session, payload_analytical))  # means get this process started and move on
                    tasks_analytical.append(task_analytical)

其功能是

async def fire_post(session, payload):

    await asyncio.sleep(1.5)
    try:
        async with session.post(url_crop, data = payload) as response:

            result_data = await response.json()



    except:
        time.sleep(3)
        await asyncio.sleep(1 / 5)
        async with session.post(url_crop, data=payload) as response:
            try:
                result_data = await response.json()
            except:
                result_data = await response.text()

    return result_data

正如你可能看到的，我有一些时间睡觉，但仍然得到一个最大的利率限制达到。我该怎么处理呢？

Answer 1

您可以使用这样的代码来限制您的请求：

import asyncio
import time


class RateLimiter:
    def __init__(self, max=1, period=1):
        self.period = period
        self.max = max
        self.signal = asyncio.Event()
        self.lock = asyncio.Lock()
        self._tasks = [asyncio.create_task(self.ticker())]
        self.signal.set()

    # This signals the event period/max times/second (so if
    # max=4 and period=1, this fires the signal ever 0.25 seconds).
    async def ticker(self):
        while True:
            await asyncio.sleep(self.period / self.max)
            self.signal.set()

    # When entering the context, 
    async def __aenter__(self):
        async with self.lock:
            await self.signal.wait()
            self.signal.clear()
        return self

    async def __aexit__(self, *args):
        pass

你就这样用它：

async def task(g):
    async with g:
        print("loop", time.ctime())


async def main():
    g = RateLimiter(max=4)
    tasks = []
    for i in range(20):
        tasks.append(asyncio.create_task(task(g)))

    await asyncio.gather(*tasks)

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

上面的输出如下所示：

loop Thu Jun  9 14:22:44 2022
loop Thu Jun  9 14:22:44 2022
loop Thu Jun  9 14:22:44 2022
loop Thu Jun  9 14:22:45 2022
loop Thu Jun  9 14:22:45 2022
loop Thu Jun  9 14:22:45 2022
loop Thu Jun  9 14:22:45 2022
loop Thu Jun  9 14:22:46 2022
loop Thu Jun  9 14:22:46 2022
loop Thu Jun  9 14:22:46 2022
loop Thu Jun  9 14:22:46 2022
loop Thu Jun  9 14:22:47 2022
...

如果查看这些时间戳，您将注意到我们从未超过4次/秒调用该print()语句(因为我们创建了一个RateLimiter(4))。