如何用gspread批量导出DataFrame至多工作表且不触发API写入限额

问题背景 #

将多个存储在字典中的DataFrame（键为工作表名，值为DataFrame）写入Google Sheets时，原循环逐个处理的方式会产生大量独立API请求：每个DataFrame需要执行创建工作表、获取工作表、写入数据、冻结首行4个操作，处理15个DataFrame就会触发60次请求，容易触及API配额限制。

原代码：

gc = gspread.oauth(credentials_filename=credential_path("credentials.json"),
                   authorized_user_filename=credential_path("token.json")
                   )

gc_sh = gc.create('rote_output_' + yyyymmddhhss())


# (rote_data 是存储DataFrame的字典)
for k, v in rote_data.items():
    gc_sh.add_worksheet(title=k, rows=100, cols=20)
    worksheet = gc_sh.worksheet(k)
    worksheet.update([v.columns.values.tolist()] + (v.fillna('')).values.tolist())
    worksheet.freeze(rows=1)

需要实现将创建工作表、写入数据、冻结首行合并为批量请求，减少API调用次数。

解决方案 #

通过gspread的batch_update()方法，可将多类操作合并为少量批量请求。核心思路是分两次批量操作：先批量创建所有工作表，再批量写入数据并冻结首行，仅需2次API请求即可完成原60次的操作。

完整实现代码 #

import gspread

gc = gspread.oauth(credentials_filename=credential_path("credentials.json"),
                   authorized_user_filename=credential_path("token.json")
                   )

# 创建目标表格
gc_sh = gc.create('rote_output_' + yyyymmddhhss())
spreadsheet_id = gc_sh.id

# 第一步：构造批量创建工作表的请求
create_requests = []
sheet_records = []
for sheet_title, df in rote_data.items():
    # 添加创建工作表的请求
    create_requests.append({
        "addSheet": {
            "properties": {
                "title": sheet_title,
                "gridProperties": {"rowCount": 100, "columnCount": 20}
            }
        }
    })
    # 暂存工作表标题和对应的数据，用于后续写入
    sheet_records.append({
        "title": sheet_title,
        "data": [df.columns.values.tolist()] + df.fillna('').values.tolist()
    })

# 执行批量创建，获取新建工作表的ID映射
create_response = gc_sh.batch_update({"requests": create_requests})
sheet_id_map = {}
for reply in create_response.get('replies', []):
    sheet_prop = reply['addSheet']['properties']
    sheet_id_map[sheet_prop['title']] = sheet_prop['sheetId']

# 第二步：构造批量写入数据和冻结首行的请求
update_requests = []
for record in sheet_records:
    sheet_id = sheet_id_map[record['title']]
    data_rows = record['data']
    total_rows = len(data_rows)
    total_cols = len(data_rows[0]) if total_rows > 0 else 0

    # 数据写入请求：使用updateCells格式
    update_requests.append({
        "updateCells": {
            "range": {
                "sheetId": sheet_id,
                "startRowIndex": 0,
                "endRowIndex": total_rows,
                "startColumnIndex": 0,
                "endColumnIndex": total_cols
            },
            "rows": [
                {"values": [{"userEnteredValue": {"stringValue": str(cell)}} for cell in row]}
                for row in data_rows
            ],
            "fields": "userEnteredValue"
        }
    })

    # 冻结首行请求：更新工作表属性
    update_requests.append({
        "updateSheetProperties": {
            "properties": {
                "sheetId": sheet_id,
                "gridProperties": {"frozenRowCount": 1}
            },
            "fields": "gridProperties.frozenRowCount"
        }
    })

# 执行批量更新
gc_sh.batch_update({"requests": update_requests})

关键说明 #

1. 分批次批量操作：先创建所有工作表并获取其ID（Google Sheets API创建工作表后会返回ID），再基于ID执行数据写入和冻结操作，确保操作顺序合法。
2. 数据格式适配：将DataFrame的行列数据转换为Google Sheets API要求的updateCells结构，每个单元格需指定userEnteredValue。
3. 请求合并效果：处理15个DataFrame仅需2次API请求，大幅降低触发配额限制的概率。

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