如何编写Python类实现REST接口JSON响应轮询并存储到CSV文件

前置依赖 #

需要先安装HTTP请求库requests，其余用到的csv、time模块均为Python标准库，无需额外安装：
pip install requests

完整代码实现 #

import requests
import csv
import time
from typing import List, Dict, Optional

class RestPoller:
    def __init__(
        self,
        api_url: str,
        poll_count: int = 3,
        poll_interval: int = 5,
        headers: Optional[Dict] = None,
        params: Optional[Dict] = None,
        csv_path: str = "poll_result.csv"
    ):
        self.api_url = api_url
        self.poll_count = poll_count
        self.poll_interval = poll_interval
        self.headers = headers or {}
        self.params = params or {}
        self.csv_path = csv_path

    def poll(self) -> List[Dict]:
        """执行轮询请求，返回所有响应JSON组成的列表"""
        result_list = []
        for i in range(self.poll_count):
            print(f"正在执行第 {i+1}/{self.poll_count} 次轮询")
            try:
                response = requests.get(
                    url=self.api_url,
                    headers=self.headers,
                    params=self.params,
                    timeout=10
                )
                response.raise_for_status() # 非200状态码直接抛出异常
                json_data = response.json()
                result_list.append(json_data)
            except Exception as e:
                print(f"第 {i+1} 次轮询失败，错误信息：{str(e)}")
            # 最后一次轮询不需要等待
            if i != self.poll_count - 1:
                time.sleep(self.poll_interval)
        return result_list

    def save_to_csv(self, data_list: List[Dict]) -> None:
        """将轮询得到的JSON列表写入CSV文件"""
        if not data_list:
            print("无有效数据可保存")
            return
        # 取第一条数据的key作为CSV表头
        headers = data_list[0].keys()
        with open(self.csv_path, mode='w', encoding='utf-8-sig', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=headers)
            writer.writeheader()
            writer.writerows(data_list)

使用示例 #

if __name__ == "__main__":
    # 替换为你的实际接口配置
    poller = RestPoller(
        api_url="https://你的实际接口地址/path",
        poll_count=5,
        poll_interval=10,
        headers={"Authorization": "Bearer 你的接口鉴权token"},
        csv_path="my_poll_result.csv"
    )
    poll_results = poller.poll()
    if poll_results:
        poller.save_to_csv(poll_results)
        print(f"共获取到{len(poll_results)}条有效数据，已保存到{poller.csv_path}")

自定义调整说明 #

• 若需要支持POST请求，可在初始化方法中新增method和payload参数，请求时根据请求方法调用对应的requests方法即可
• 若接口返回的JSON为嵌套结构，可新增flatten_json工具方法对单条数据做打平处理后再写入CSV，避免嵌套字段被存为字符串
• 若需要对请求失败的场景做重试，可在异常捕获逻辑中增加重试计数，避免单次请求失败占用轮询次数

内容的提问来源于stack exchange，提问作者M-Programmer