如何减慢异步 API 调用以匹配 API 限制？

Question

我有一个 ~300K 的 URL 列表，用于 API 我需要从中获取数据。

API 限制为每秒 100 次调用。

我已经为异步创建了一个 class，但这工作速度很快，我在 API 上遇到了错误。

如何降低异步速度，以便每秒调用 100 次？

import grequests

lst = ['url.com','url2.com']

class Test:
    def __init__(self):
        self.urls = lst

    def exception(self, request, exception):
        print ("Problem: {}: {}".format(request.url, exception))

    def async(self):
        return grequests.map((grequests.get(u) for u in self.urls), exception_handler=self.exception, size=5)



    def collate_responses(self, results):
        return [x.text for x in results]

test = Test()
#here we collect the results returned by the async function
results = test.async()
response_text = test.collate_responses(results)

Answer 1

This SO 中的其他问题显示了如何执行此操作。对了，你需要的通常叫做throttling.

Answer 2

您可以只记录已经过去了多少时间，然后决定是否要执行更多请求。

这将每秒打印 100 个数字，例如：

from datetime import datetime
import time

start = datetime.now()
time.sleep(1);
counter = 0
while (True):
    end = datetime.now()
    s = (end-start).seconds
    if (counter >= 100):
        if (s <= 1):
            time.sleep(1) # You can keep track of the time and sleep less, actually
            start = datetime.now()
            counter = 0
    print(counter)
    counter += 1

Answer 3

我采取的第一步是创建一个对象，该对象每 t 毫秒最多可以分配 n 个硬币。

import time

class CoinsDistribution:
    """Object that distribute a maximum of maxCoins every timeLimit ms"""
    def __init__(self, maxCoins, timeLimit):
        self.maxCoins = maxCoins
        self.timeLimit = timeLimit
        self.coin = maxCoins
        self.time = time.perf_counter()


    def getCoin(self):
        if self.coin <= 0 and not self.restock():
            return False

        self.coin -= 1
        return True

    def restock(self):
        t = time.perf_counter()
        if (t - self.time) * 1000 < self.timeLimit:
            return False
        self.coin = self.maxCoins
        self.time = t
        return True

现在我们需要一种方法来强制函数仅在他们可以获得硬币时才被调用。为此，我们可以编写一个可以像这样使用的装饰器函数：

@limitCalls(callLimit=1, timeLimit=1000)
def uniqFunctionRequestingServer1():
    return 'response from s1'

但有时，多个函数调用请求同一个服务器，因此我们希望它们从同一个 CoinsDistribution 对象获取硬币。因此，装饰器的另一种用途是提供 CoinsDistribution 对象：

server_2_limit = CoinsDistribution(3, 1000)

@limitCalls(server_2_limit)
def sendRequestToServer2():
    return 'it worked !!'

@limitCalls(server_2_limit)
def sendAnOtherRequestToServer2():
    return 'it worked too !!'

我们现在必须创建装饰器，它可以使用 CoinsDistribution 对象或足够的数据来创建一个新装饰器。

import functools

def limitCalls(obj=None, *, callLimit=100, timeLimit=1000):
    if obj is None:
        obj = CoinsDistribution(callLimit, timeLimit)

    def limit_decorator(func):
        @functools.wraps(func)
        def limit_wrapper(*args, **kwargs):
            if obj.getCoin():
                return func(*args, **kwargs)
            return 'limit reached, please wait'
        return limit_wrapper
    return limit_decorator

大功告成！现在，您可以限制调用任何 API 的次数，如果您必须管理大量 CoinsDistribution 对象（到不同的 API 端点或不同 APIs).

注意：这里我选择了 return 如果没有硬币可用则显示错误消息。您应该根据您的需要调整此行为。

如何减慢异步 API 调用以匹配 API 限制？

How to slow down asynchrounous API calls to match API limits?

python

asynchronous

api-design

python-3.x

grequests