如果您提供来自 API 的示例响应，将会很有帮助。
如果 API 配置正确，在给定响应中将有一个 next 属性，可引导您进入下一页。
然后，您可以使用next 中给出的链接递归地继续调用API。最后一页的Link header中不会有next。
resp.links["next"]["url"] 将为您提供下一页的 URL。
例如，GitHub API 具有 next、last、first 和 prev 属性。
要把它变成代码，首先你需要把你的代码变成函数。
鉴于每页最多有 500 个结果，这意味着您正在从 API 中提取某种类型的数据列表。通常，这些数据会在raw 内的某个列表中返回。
现在，假设您要提取raw.get('data') 列表中的所有元素。
import requests
header = {"Authorization": "Bearer <my_api_token>"}
results_per_page = 500
def compose_url():
return (
"https://my_api.com/v1/users"
+ "?per_page="
+ str(results_per_page)
+ "&page_number="
+ "1"
)
def get_result(url=None):
if url_get is None:
url_get = compose_url()
else:
url_get = url
s = requests.Session()
s.proxies = {"http": "<my_proxies>", "https": "<my_proxies>"}
resp = s.get(url_get, headers=header, verify=False)
# You may also want to check the status code
if resp.status_code != 200:
raise Exception(resp.status_code)
raw = resp.json() # of type dict
data = raw.get("data") # of type list
if not "url" in resp.links.get("next"):
# We are at the last page, return data
return data
# Otherwise, recursively get results from the next url
return data + get_result(resp.links["next"]["url"]) # concat lists
def main():
# Driver function
data = get_result()
# Then you can print the data or save it to a file
if __name__ == "__main__":
# Now run the driver function
main()
但是，如果没有正确的 Link 标头，我会看到 2 个解决方案：
(1) 递归和 (2) 循环。
我将演示递归。
正如您所提到的，当 API 响应中存在分页时，即每页的最大结果数有限制时，通常会有一个名为 page number 或 start index 的查询参数来指示哪个您正在查询的“页面”，因此我们将在代码中使用page_number 参数。
逻辑是：
- 给定一个 HTTP 请求响应，如果少于 500 个结果，则意味着没有更多页面。返回结果。
- 如果给定响应中有 500 个结果，则意味着可能还有另一个页面，因此我们将
page_number 前进 1 并执行递归（通过调用函数本身）并与之前的结果连接。
import requests
header = {"Authorization": "Bearer <my_api_token>"}
results_per_page = 500
def compose_url(results_per_page, current_page_number):
return (
"https://my_api.com/v1/users"
+ "?per_page="
+ str(results_per_page)
+ "&page_number="
+ str(current_page_number)
)
def get_result(current_page_number):
s = requests.Session()
s.proxies = {"http": "<my_proxies>", "https": "<my_proxies>"}
url = compose_url(results_per_page, current_page_number)
resp = s.get(url, headers=header, verify=False)
# You may also want to check the status code
if resp.status_code != 200:
raise Exception(resp.status_code)
raw = resp.json() # of type dict
data = raw.get("data") # of type list
# If the length of data is smaller than results_per_page (500 of them),
# that means there is no more pages
if len(data) < results_per_page:
return data
# Otherwise, advance the page number and do a recursion
return data + get_result(current_page_number + 1) # concat lists
def main():
# Driver function
data = get_result(1)
# Then you can print the data or save it to a file
if __name__ == "__main__":
# Now run the driver function
main()
如果您真的想存储raw 响应，您可以。但是，您仍然需要检查给定响应中的结果数。逻辑类似。如果给定的raw 包含 500 个结果，则意味着可能还有另一个页面。我们将页码前移 1 并进行递归。
我们仍然假设raw.get('data') 是长度为结果数的列表。
因为 JSON/字典文件不能简单地串联，所以可以将每个页面的raw（即dictionary）存储到raws 的列表中。然后，您可以以任何您想要的方式解析和合成数据。
使用以下get_result函数：
def get_result(current_page_number):
s = requests.Session()
s.proxies = {"http": "<my_proxies>", "https": "<my_proxies>"}
url = compose_url(results_per_page, current_page_number)
resp = s.get(url, headers=header, verify=False)
# You may also want to check the status code
if resp.status_code != 200:
raise Exception(resp.status_code)
raw = resp.json() # of type dict
data = raw.get("data") # of type list
if len(data) == results_per_page:
return [raw] + get_result(current_page_number + 1) # concat lists
return [raw] # convert raw into a list object on the fly
至于循环方法，逻辑类似于递归。本质上，您将多次调用get_result() 函数，收集结果，并在最远的页面包含少于500 个结果时尽早break。
如果您事先知道结果总数，您可以简单地运行循环预定次数。
你关注吗？您还有其他问题吗？
（我对“将其加载到 JSON 文件中”的意思有点困惑。您的意思是将最终的 raw 结果保存到 JSON 文件中吗？或者您指的是 .json() 中的方法resp.json()？在这种情况下，您不需要import json 来执行resp.json()。resp 上的.json() 方法实际上是requests 模块的一部分。
另外一点，您可以使您的 HTTP 请求异步，但这稍微超出了您最初问题的范围。
附：我很高兴了解人们使用的其他解决方案，也许是更优雅的解决方案。