如何终止Python API请求的while循环？基于totalPages倒计时方案

核心问题分析 #

你的代码存在几个关键问题导致循环无法终止：

• cursor未作用于请求：url2在循环外定义，即使nextKey更新，请求仍使用初始的空cursor URL
• cursor判断逻辑缺失：仅检查"cursor"字段存在，但未判断该cursor是否和上一次重复（API返回重复cursor代表无更多数据）
• totalpage变量错误：totalpage = resp["metadata"]取的是整个metadata对象，而非具体的分页数值
• Session滥用：循环内每次创建新Session，未复用连接，且重试配置无效

最优终止方案 #

方案1：基于cursor重复检测（推荐） #

API返回的cursor如果和上一次请求的cursor相同，说明已经到最后一页，直接终止循环。这是最可靠的方式，适配绝大多数分页API。

方案2：基于已获取资源数与总资源数对比 #

通过metadata.totalResources获取总资源数，每次累加当前页的资源数量，当已获取数量≥总资源数时终止。

方案3：基于totalPages计数 #

如果API返回metadata.totalPages，可以初始化页码计数器，每请求一次递增，达到总页数时终止。

修复后的完整代码 #

import requests
import sys
import urllib3
import json
from requests import Session
from requests.adapters import HTTPAdapter, Retry

# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
sys.path.append('/Users/s153152/Documents/Workspace/kpi/rapid7')

def _headers():
    return {
        'X-Api-Key': 'key',
        'Content-type': 'application/json', 
        'Accept': '*/*'
    }

def _request():
    third_party_patching_filter = {
        "asset": "asset.tags IN ['osswin'] && asset.os.vendor CONTAINS 'microsoft' && asset.agentKey IS NOT NULL", 
        "vulnerability": "vulnerability.categories NOT IN ['microsoft patch']"
    }

    headers = _headers()
    base_url = "https://us.api.insight.rapid7.com/vm/v4/integration/assets"
    results = []
    
    # 初始化Session和重试配置（复用连接）
    session = Session()
    retries = Retry(
        total=10,
        backoff_factor=0.3,
        status_forcelist=[502, 503, 504]
    )
    session.mount(base_url, HTTPAdapter(max_retries=retries))
    session.verify = False

    next_cursor = ""
    previous_cursor = None
    total_resources = 0
    fetched_count = 0

    while True:
        # 每次请求动态拼接cursor参数
        params = {'size': 50}
        if next_cursor:
            params['cursor'] = next_cursor
        
        resp = session.post(
            url=base_url,
            headers=headers,
            json=third_party_patching_filter,
            params=params
        )
        resp.raise_for_status()  # 抛出HTTP错误，便于排查
        resp_data = resp.json()

        # 首次请求获取总资源数
        if fetched_count == 0:
            total_resources = resp_data["metadata"].get("totalResources", 0)
            print(f"总资源数: {total_resources}")

        # 处理当前页数据
        for asset in resp_data["data"]:
            # 检查是否包含OSSWIN标签
            has_osswin = any(tag["name"] == 'OSSWIN' for tag in asset.get('tags', []))
            if not has_osswin:
                continue
            
            try:
                results.append({
                    'ip_address': asset['ip'],
                    'os_name': asset['host_name'],
                    'critical_vuln': asset['critical_vulnerabilities'],
                    'os_type': asset['os_type']
                })
            except KeyError as e:
                print(f"资产字段缺失: {e}, IP: {asset.get('ip', '未知')}")
        
        fetched_count += len(resp_data["data"])
        print(f"已获取 {fetched_count}/{total_resources} 条数据")

        # 处理cursor，判断是否终止循环
        current_cursor = resp_data["metadata"].get("cursor")
        # 终止条件：cursor为空 或 和上一次重复 或 已获取全部资源
        if not current_cursor or current_cursor == previous_cursor or fetched_count >= total_resources:
            break
        
        previous_cursor = current_cursor
        next_cursor = current_cursor

    # 循环结束后统一写入JSON，避免重复IO
    with open('server_info.json', 'w') as f:
        json.dump(results, f, indent=2)
    
    print(f"数据采集完成，共获取 {len(results)} 条有效资产数据")
    return [item['ip_address'] for item in results], len(results)

def smart_filter():
    ips, count = _request()
    print(f"采集到的IP列表: {ips}")
    print(f"总IP数: {count}")

smart_filter()

关键修复点说明 #

1. 复用Session：循环外创建Session并配置重试，提升请求效率和稳定性
2. 动态拼接URL参数：每次请求根据next_cursor动态生成参数，确保cursor生效
3. 多重终止条件：同时检查cursor重复、cursor为空、已获取数量≥总资源数，避免无限循环
4. 优化数据处理：用any()简化标签判断，统一处理字段缺失异常
5. 减少IO操作：循环结束后一次性写入JSON，提升性能

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