我有一个看起来像这样的列表:
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