Python分页读取REST API数据异常排查（每页最大100条）

问题背景 #

尝试用Python从Airtable的REST API提取近4000条数据并保存为JSON文件，但API限制每页最多返回100条记录。编写的分页请求代码最终仅返回['records']，无法正确收集数据。

原代码 #

import requests
import json
from requests.structures import CaseInsensitiveDict

url = "https://api.airtable.com/v0/CENSORED/Vendors?maxRecords=100"

headers = CaseInsensitiveDict()
headers["Authorization"] = "Bearer CENSORED"

resp = requests.get(url, headers=headers)

resp.content.decode("utf-8")

vendors = []
new_results = True
page = 1
while new_results:
    centiblock = requests.get(url + f"&page={page}", headers=headers).json()
    new_results = centiblock.get("results", [])
    vendors.extend(centiblock)
    page += 1

full_directory = json.dumps(vendors, indent=4)

print(full_directory)

API单页返回格式示例 #

{
    "records": [
        {
            "id": "XXX",
            "createdTime": "2018-10-15T19:23:59.000Z",
            "fields": {
                "Vendor Name": "XXX",
                "Main Phone": "XXX",
                "Street": "XXX",
                ...
            }
        }
    ]
}

错误原因分析 #

1. 分页参数不符合API规范：Airtable API并不支持page参数分页，正确的分页方式是通过响应中的offset字段获取下一页数据，原代码的分页逻辑完全无效。
2. 数据提取逻辑错误：API的实际记录数据存放在records字段中，但原代码直接对响应字典执行extend操作，这会将字典的键（即'records'）添加到列表中，而非实际记录数据。
3. 终止条件错误：原代码用centiblock.get("results", [])判断是否有新数据，但API返回的字段是records而非results，导致终止条件逻辑失效。

修正后的代码 #

import requests
import json
from requests.structures import CaseInsensitiveDict

base_url = "https://api.airtable.com/v0/CENSORED/Vendors"
headers = CaseInsensitiveDict()
headers["Authorization"] = "Bearer CENSORED"

vendors = []
offset = None

while True:
    # 构建请求参数
    params = {"maxRecords": 100}
    if offset:
        params["offset"] = offset
    
    # 发送请求并处理错误
    resp = requests.get(base_url, headers=headers, params=params)
    resp.raise_for_status()
    data = resp.json()
    
    # 提取并添加记录数据
    records = data.get("records", [])
    if not records:
        break
    vendors.extend(records)
    
    # 更新分页偏移量，无偏移量则终止循环
    offset = data.get("offset")
    if not offset:
        break

# 保存数据到JSON文件
with open("vendors.json", "w", encoding="utf-8") as f:
    json.dump(vendors, f, indent=4, ensure_ascii=False)

print(f"共收集到{len(vendors)}条数据")

关键修正说明 #

1. 使用官方分页参数：改用Airtable API指定的offset参数实现分页，确保每次请求能获取下一页数据。
2. 正确提取记录：从响应的records字段中取出实际记录列表，通过extend添加到结果列表中。
3. 合理终止循环：当响应中无记录或无offset时终止循环，确保所有数据被完整收集。
4. 添加错误处理：通过resp.raise_for_status()捕获HTTP请求错误，便于调试。
5. 直接写入文件：使用json.dump直接将数据写入文件，避免不必要的字符串转换。

内容的提问来源于stack exchange，提问作者Lif