如何高效遍历分页API获取全部数据：以德州房产信息查询为例-Python教程-PHP中文网

如何高效遍历分页API获取全部数据：以德州房产信息查询为例

本文详解如何通过循环请求分页api（如wcad房产搜索接口）获取全部结果，涵盖自动翻页逻辑、错误重试机制、请求节流与资源管理等关键实践。

本文详解如何通过循环请求分页api（如wcad房产搜索接口）获取全部结果，涵盖自动翻页逻辑、错误重试机制、请求节流与资源管理等关键实践。

在开发类似“德州人名地址查询”这类数据聚合型应用时，核心挑战之一是可靠地遍历后端返回的分页数据集。许多公共政务API（如威尔县税务评估系统 WCAD）采用基于页码（pn）的分页策略，但不直接返回总页数——而是通过每次响应中的 RecordCount 字段判断是否还有下一页：当该值为 0 时，即表示当前页为末页或无更多结果。

以下是一个健壮、生产就绪的分页遍历实现方案：

讯飞绘文

讯飞绘文：免费AI写作/AI生成文章

下载

✅ 推荐实现：基于 RecordCount 的自适应翻页

import requests
from requests.adapters import HTTPAdapter, Retry

URL = "https://search.wcad.org/ProxyT/Search/Properties/quick/"

def search_by_name(name: str, max_retries: int = 5):
    params = {
        "f": name,
        "pn": 1,          # 初始页码（从1开始）
        "st": 4,          # 排序字段（示例中为OwnerName）
        "so": "desc",     # 排序方向
        "pt": "RP;PP;MH;NR",  # 财产类型
        "ty": "2024"      # 税务年份
    }

    # 配置重试策略：对 429（Too Many Requests）等临时错误自动重试
    retry_strategy = Retry(
        total=max_retries,
        status_forcelist=[429, 500, 502, 503, 504],
        backoff_factor=1.0
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)

    with requests.Session() as session:
        session.mount("https://", adapter)  # 仅对HTTPS启用重试

        while True:
            try:
                response = session.get(URL, params=params, timeout=10)
                response.raise_for_status()  # 抛出HTTP错误（如404/500）

                data = response.json()
                record_count = data.get("RecordCount", 0)

                if record_count == 0:
                    print("✅ 遍历完成：无更多记录。")
                    break

                print(f"? 第 {params['pn']} 页：获取到 {record_count} 条记录")

                # ✨ 此处可安全解析并处理 ResultList 数据（如提取 OwnerName + SitusAddress）
                # 示例：
                # for item in data.get("ResultList", []):
                #     print(f"{item.get('OwnerName', 'N/A')} — {item.get('SitusAddress', 'N/A')}")

                params["pn"] += 1  # 自增页码，进入下一轮请求

                # ⚠️ 强烈建议添加合理延迟（如 0.5–2 秒），避免触发反爬或服务限流
                import time
                time.sleep(1.0)

            except requests.exceptions.RequestException as e:
                print(f"❌ 请求失败（第 {params['pn']} 页）：{e}")
                break
            except ValueError as e:
                print(f"❌ JSON解析失败：{e}")
                break

# 使用示例
search_by_name("Kennedy")