如何基于CSV数据构建API循环请求并动态生成JSON Payload

需求说明 #

从CSV文件读取数据并循环发起API请求：

• 每行数据对应一个请求，URL中仅使用该行的USER_ID
• 对该行所有非空的BRANCH_ID列（如BRANCH_ID1、BRANCH_ID2等），生成格式为{"id": BRANCH_ID值, "with_descendants": True}的节点
• 所有节点组成数组，作为请求体中的node_list

原始代码问题 #

• 仅读取前两列数据，无法处理多列BRANCH_ID
• 请求体的node_list写死为固定数量节点，未过滤空值
• 变量替换错误（USER_ID、BRANCH_ID1等未正确引用）
• 缺少请求头定义，代码缩进逻辑错误

修正后的代码 #

import http.client
import json
import csv

# 定义API请求头
headers = {
    'Content-Type': 'application/json'
}

with open('sample.csv', 'r') as file:
    # 用DictReader处理可变列数的CSV，自动识别表头
    reader = csv.DictReader(file)
    for row in reader:
        user_id = row['USER_ID']
        node_list = []
        
        # 遍历所有BRANCH_ID列，仅保留非空值的节点
        for col_name, value in row.items():
            if col_name.startswith('BRANCH_ID') and value.strip():
                node_list.append({
                    "id": value,
                    "with_descendants": True
                })
        
        # 无有效BRANCH_ID时跳过请求
        if not node_list:
            print(f"用户{user_id}无有效分支ID，跳过本次请求")
            continue
        
        # 构建请求体
        payload = json.dumps({
            "node_list": node_list
        })
        
        # 发起PUT请求
        conn = http.client.HTTPSConnection("www.api.com")
        request_url = f"/poweruser/v1/powerusers/{user_id}/branches?access_token=0ae4f75d988900902dd9bfaa2df048e80a6fb30b"
        conn.request("PUT", request_url, payload, headers)
        
        # 处理响应
        res = conn.getresponse()
        data = res.read()
        print(f"用户{user_id}请求响应：{data.decode('utf-8')}")
        conn.close()

关键改进点 #

• 使用csv.DictReader：无需预先指定列数，自动适配任意数量的BRANCH_ID列
• 动态生成node_list：遍历所有以BRANCH_ID开头的列，过滤空值后生成节点
• 修复变量替换：正确将USER_ID插入请求URL，避免硬编码错误
• 添加请求头：明确指定Content-Type为application/json，符合API要求
• 增加容错逻辑：无有效BRANCH_ID时跳过请求，避免无效调用
• 完善资源管理：每次请求后关闭连接，防止资源泄漏

示例CSV（sample.csv） #

USER_ID,BRANCH_ID1,BRANCH_ID2,BRANCH_ID3,BRANCH_ID4,BRANCH_ID5
14411,332,441,446,445,110
15512,445,221,224
24412,567
23211,955,224

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