时间戳

首页 > 解决方案 > 有页面限制时如何转储 API 请求的所有结果？

json - 有页面限制时如何转储 API 请求的所有结果？

问题描述

我正在使用 API 从 url 中提取数据，但是 API 有分页限制。它是这样的：

我有一个脚本，我可以得到一个页面或每个页面的结果，但我想自动化它。我希望能够遍历所有页面或 per_page(500) 并将其加载到 json 文件中。

这是我的代码，每页可以得到 500 个结果：

import json, pprint
import requests

url = "https://my_api.com/v1/users?per_page=500"
header = {"Authorization": "Bearer <my_api_token>"}

s = requests.Session()
s.proxies = {"http": "<my_proxies>", "https": "<my_proxies>" }

resp = s.get(url, headers=header, verify=False)
raw=resp.json()
for x in raw:
    print(x)

输出为 500，但有没有办法继续并从停止的地方开始提取结果？或者甚至逐页获取每页的所有数据，直到页面中没有数据？

标签： jsonpython-3.xapihttpspython-requests

解决方案

如果您提供来自 API 的示例响应，这将很有帮助。

如果 API 配备正确，则next给定响应中将有一个属性将您引导至下一页。

next然后，您可以使用递归中给出的链接继续调用 API 。在最后一页，链接标题中将没有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 numberorstart index的查询参数来指示您正在查询哪个“页面”，所以我们'将利用page_number代码中的参数。

逻辑是：

  • 给定一个 HTTP 请求响应，如果结果少于 500，则意味着没有更多页面。返回结果。
  • 如果给定响应中有 500 个结果，则意味着可能还有另一个页面，因此我们前进page_number1 并执行递归（通过调用函数本身）并与先前的结果连接。
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（即 a ）存储到s 列表中。然后，您可以以任何您想要的方式解析和合成数据。dictionaryraw

使用以下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()多次调用该函数，收集结果，并break在最远的页面包含少于 500 个结果的早期。

如果您事先知道结果总数，您可以简单地运行循环预定次数。

你跟吗？您还有其他问题吗？

（我对“将其加载到 JSON 文件中”的意思有点困惑。您的意思是将最终raw结果保存到 JSON 文件中吗？还是您指的是中的.json()方法resp.json()？在这种情况下，您不会需要import json做resp.json(). .json()onresp方法实际上是requests模块的一部分。

额外的一点是，您可以使您的 HTTP 请求异步，但这稍微超出了您最初问题的范围。

Ps 我很高兴了解人们使用的其他解决方案，也许是更优雅的解决方案。

推荐阅读