Python基于CSV按CIN匹配构造Airtable PATCH更新请求咨询

首先纠正两个关键问题：

1. 你之前写的jsondata = json.loads(json.dumps(data, indent=4))是完全冗余的操作，csv.DictReader读取出来的行本身就是标准字典结构，可以直接使用，不需要做两次JSON序列化/反序列化。
2. Airtable官方PATCH更新接口不支持直接通过自定义字段（比如你的CIN）做匹配更新，你之前写的请求体模板漏了必填的记录id参数——Airtable所有单条记录的增删改都依赖系统生成的rec开头的唯一记录ID，直接把CIN传到fields里接口无法识别要更新的目标记录。

完整实现步骤 #

1. 先拉取Airtable表内CIN与记录ID的映射关系 #

要按CIN匹配更新，首先得把表内现有CIN和对应记录ID的对应关系拉到本地。注意Airtable单次查询最多返回100条记录，记录量大时需要自动翻页：

import csv
import os
import requests

endpoint = f"https://api.airtable.com/v0/{os.environ['database']}"
headers = {
    "Authorization": f"Bearer {os.environ['airtable_key']}",
    "Content-Type": "application/json",
}

def get_cin_to_id_map(table_name):
    cin_record_map = {}
    offset = None
    while True:
        # 只查CIN字段，减少请求传输量
        req_params = {"fields[]": ["CIN"], "offset": offset}
        resp = requests.get(
            f"{endpoint}/{table_name}",
            headers=headers,
            params=req_params
        ).json()
        # 构建CIN到记录ID的映射
        for record in resp["records"]:
            if "CIN" in record["fields"]:
                cin_record_map[record["fields"]["CIN"].strip()] = record["id"]
        # 无下一页时退出循环
        if "offset" not in resp:
            break
        offset = resp["offset"]
    return cin_record_map

2. 读取CSV组装合法请求体 #

你构思的第二种思路（先构建records列表再塞入请求体）是更稳妥的实现方式，注意Airtable单次PATCH请求最多支持更新10条记录，CSV记录数较多时需要做分批处理：

def build_update_batches(csv_file_path, cin_record_map):
    all_update_records = []
    # 读CSV时指定utf-8编码，避免法语特殊字符乱码
    with open(csv_file_path, "rt", encoding="utf-8") as csvf:
        csv_reader = csv.DictReader(csvf)
        for row in csv_reader:
            cin = row["CIN"].strip()
            # 跳过Airtable中不存在对应CIN的行，避免接口报错
            if cin not in cin_record_map:
                print(f"跳过未匹配到记录的CIN：{cin}")
                continue
            # CSV读取的内容都是字符串，需要转成整数匹配Airtable数字字段类型
            work_days = int(row["Nombre de jours travaillés"].strip())
            all_update_records.append({
                "id": cin_record_map[cin],
                # PATCH是增量更新，只需要传要修改的字段即可，不需要传CIN或其他无关字段
                "fields": {
                    "Nombre de jours travaillés": work_days
                }
            })
    # 按每10条切分，适配接口单次请求上限
    return [all_update_records[i:i+10] for i in range(0, len(all_update_records), 10)]

3. 复用你的PATCH方法发送请求 #

def patch(table, data):
    url = f"{endpoint}/{table}"
    resp = requests.patch(url=url, json=data, headers=headers)
    # 加基础错误提示，方便排查问题
    if resp.status_code != 200:
        print(f"更新请求报错，状态码{resp.status_code}，返回内容：{resp.text}")
    return resp.json()

# 主逻辑调用
if __name__ == "__main__":
    target_table_name = "替换成你的Airtable表名"
    csv_path = "替换成你的CSV文件路径.csv"
    # 获取映射关系
    cin_id_map = get_cin_to_id_map(target_table_name)
    # 构建分批更新数据
    update_batches = build_update_batches(csv_path, cin_id_map)
    # 逐批发送更新请求
    for batch in update_batches:
        req_body = {"records": batch}
        patch_result = patch(target_table_name, req_body)
        print(f"批次更新完成，共处理{len(batch)}条记录")

为什么不推荐思路a #

如果提前创建固定长度的模板字典，一旦出现CSV行对应CIN在Airtable不存在、CSV行数和初始化模板长度不匹配的情况，很容易触发索引越界、空值覆盖已有数据的问题，用动态append构建列表的方式容错性高很多，也更容易做异常数据过滤。

内容的提问来源于stack exchange，提问作者Kaiss B.