我正在尝试编写一个函数 my_func,它从 1 页的 API 获取数据。 my_func 在另一个函数的循环中被调用。虽然它成功运行,但我很难确定我的错误处理是否正确。
问题是,我正在循环浏览 2000 多个页面。所以,我不希望 1 页错误导致一切失败。当我的代码失败时,我将其设置为回滚更改。更改是将新数据添加到 Redshift 表中。
import requests as r
import time
def my_func(api_key, page):
base_url = f'https://example.com/api/?limit=1000&page={page}'
headers = {'Authorization': f"{api_key}", 'Content-Type': 'application/json'}
try:
response = r.get(url=base_url, headers=headers)
except Exception as e:
print(f"Failed to get data from API: {e}")
else:
#If unsuccessful API call, sleep 2 seconds and try again
if response.status_code != 200:
time.sleep(2)
response = r.get(url=base_url, headers=headers)
#If 2nd unsuccessful API call, print status message and pass
if response.status_code != 200:
print(f"Status code {response.status_code} \n Failed to get data from API on page {page}")
pass #return None, None
#If 2nd API call is successful, save data to dictionary as a tuple
else:
dict1 = response.json()['dict1']
dict2 = response.json()['dict2']
return dict1, dict2
#If successful API call, save data to dictionary as a tuple
else:
dict1 = response.json()['dict1']
dict2 = response.json()['dict2']
return dict1, dict2
if response.status_code != 200:
print(f"Status code {response.status_code} \n Failed to get data from API on page {page}")
pass #return None, None
我只需要知道这段代码是否按照我认为应该的方式读取和执行。
尝试得到回应
例外:打印消息
无例外:打印消息
如果 API 调用错误:睡眠并重试 如果第二个 API 调用错误:打印消息并传递(我应该传递还是返回 None、None?)
否则第二个 API 调用很好:将字典保存为元组并返回元组
其他好的 API 调用:将字典保存为元组并返回元组
你可以尝试这个库坚韧。它可以解决您的问题,此外还可以解决更多问题,例如增加重试次数、根据异常或结果重试等