如何在将列表中的每个值传递到 API 并在每个列表列表之后暂停时迭代列表列表?



我有一个看起来像这样的列表:

lst = [1,2,3, etc]

我可以成功迭代此列表并将 ID 传递到 API 中。但是,API 在我传递的 Id 数量方面在某个点后中断。

我继续使用以下代码将列表分成 50 个块:

def chunks(l, n):
# For item i in a range that is a length of l,
for i in range(0, len(l), n):
# Create an index range for l of n items:
yield l[i:i+n]
a = list(chunks(lst, 50))

a变量是列表列表,如下所示:

[[1,2,3][4,5,6]]

如何遍历列表列表,将单个 Id 传递到 API 代码(如下(,同时在每个块后暂停代码 10 秒?

此外,如果 API 中断,我是否可以强制 API 继续其余代码?目前不是必需的,因此这是可选的。

接口代码:

lst1=[]
for i in lst:
#print (row)
url = 'url/Id={}'.format(i)
r = requests.get(url).text
lst1.append(xml_data1)
#time.sleep(1)
print(xml_data1)

提前谢谢你。

您可以在执行内部循环后使用sleep函数使用双for循环。下面是一个示例:

lst1 = []
for chunk in a:
for id in chunk:
url = 'url/Id={}'.format(id)
xml_data = requests.get(url).text
lst1.append(xml_data)
sleep(10) # will pause execution for 10 seconds

要回答第二个问题,即在 API 引发错误时如何继续执行,您可以使用try/except。可能引发错误的代码位于try块中,如果引发错误,则执行将移动到指定如何继续的except块。如果您将代码包装在函数中,则可以使这变得更容易一些。

我发现跟踪您尝试使用可用于指数退避的关键字参数调用特定id的次数很有帮助 - 根据您尝试的次数,您等待更长的时间。

下面是一个基于上述代码构建的示例:

def get_data_on_id(id, try_num=1):
url = 'url/Id={}'.format(id)
try:
return requests.get(url).text
except:
sleep(2**try_num) # exponential backoff
return get_data_on_id(id, try_num=try_num + 1)
lst1 = []
for chunk in a:
for id in chunk:
xml_data = get_data_on_id(id)
lst1.append(xml_data)
sleep(10)

上面的代码与它之前的代码大致相同,但请求逻辑包装在包含try/except块的get_data_on_id函数中。如果在请求url时抛出错误,该函数将等待2^try_num秒,然后重试。

您可能还希望except特定的错误并以不同的方式处理它们(或者只是为了明确,以便您只except您准备处理的错误(。以下是try/except上的文档: https://docs.python.org/3/tutorial/errors.html

相关内容

最新更新