如何批量向HubSpot API推送大JSON文件数据？

核心实现逻辑 #

• 读取大JSON文件，按15000条记录拆分批次
• 每批数据按HubSpot API要求格式化（包装为{"inputs": [...]}结构）
• 发送POST请求，捕获HTTP错误、SSL错误等异常
• 每批请求完成后强制休眠10秒，避免触发API限流
• 记录每批次的推送结果（成功/失败、响应信息）

完整代码实现 #

import json
import requests
import time

def batch_push_to_hubspot(json_file_path, hubspot_api_url, api_token, batch_size=15000, sleep_time=10):
    # 读取JSON数据
    with open(json_file_path, 'r', encoding='utf-8') as f:
        all_records = json.load(f)
    
    total_records = len(all_records)
    print(f"总记录数：{total_records}，将分为{((total_records + batch_size - 1) // batch_size)}批次推送")

    # 设置请求头
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json"
    }

    # 分批处理
    for batch_idx in range(0, total_records, batch_size):
        batch_data = all_records[batch_idx:batch_idx + batch_size]
        current_batch_num = (batch_idx // batch_size) + 1
        print(f"开始推送第{current_batch_num}批次，共{len(batch_data)}条记录")

        # 按HubSpot API要求格式化数据
        payload = {"inputs": batch_data}

        try:
            # 发送请求，可根据情况处理SSL验证
            response = requests.post(
                hubspot_api_url,
                headers=headers,
                json=payload
                # 若遇SSLError，可临时关闭SSL验证（仅测试环境使用）
                # verify=False
            )
            # 主动触发HTTP错误抛出
            response.raise_for_status()
            print(f"第{current_batch_num}批次推送成功，响应：{response.json()}")
        except requests.exceptions.HTTPError as e:
            print(f"第{current_batch_num}批次推送失败（HTTP错误）：{e.response.status_code} - {e.response.text}")
        except requests.exceptions.SSLError as e:
            print(f"第{current_batch_num}批次推送失败（SSL错误）：{str(e)}")
            # 尝试重新发送并关闭SSL验证
            try:
                response = requests.post(
                    hubspot_api_url,
                    headers=headers,
                    json=payload,
                    verify=False
                )
                response.raise_for_status()
                print(f"重新推送第{current_batch_num}批次成功（跳过SSL验证）")
            except Exception as retry_e:
                print(f"重新推送第{current_batch_num}批次仍失败：{str(retry_e)}")
        except Exception as e:
            print(f"第{current_batch_num}批次推送失败（未知错误）：{str(e)}")
        
        # 休眠10秒，最后一批次可省略
        if batch_idx + batch_size < total_records:
            print(f"休眠{sleep_time}秒...")
            time.sleep(sleep_time)

# 使用示例
if __name__ == "__main__":
    # 替换为你的实际参数
    JSON_FILE = "your_large_data.json"
    HUBSPOT_URL = "https://api.hubapi.com/crm/v3/objects/contacts/batch/create"  # 以联系人批量创建为例
    API_TOKEN = "your_hubspot_api_token"
    
    batch_push_to_hubspot(JSON_FILE, HUBSPOT_URL, API_TOKEN)

关键细节说明 #

解决400 Bad Request问题 #

• 数据格式验证：HubSpot批量API要求请求体必须是{"inputs": [单个记录对象]}结构，确保每条记录包含email和properties字段，且字段格式符合规范（比如email必须是合法邮箱格式）
• 请求头检查：必须携带Authorization（Bearer token）和Content-Type: application/json，缺少或格式错误会直接返回400
• 单条数据预处理：提前遍历数据，过滤掉空email、properties缺失等格式错误的记录，避免整批失败

解决SSLError问题 #

• 测试环境临时方案：添加verify=False跳过SSL验证，但生产环境禁止使用
• 生产环境合规方案：下载并指定HubSpot的CA证书路径，或确保运行环境已信任对应CA证书，示例：verify="/path/to/ca_cert.pem"

其他注意事项 #

• 限流适配：除每批休眠10秒，可读取API返回的X-RateLimit-Remaining头信息，若剩余调用次数不足，动态延长休眠时间
• 断点续推：可添加记录已推送批次的逻辑（比如写入本地文件），避免程序中断后重复推送全部数据

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