Python脚本调用Google Sheets API遭遇写入配额超限问题咨询

问题重现 #

我编写了以下Python函数，意图每3秒更新一次Google表格：

def update_gsheet():
    global ce_trade_status, pe_trade_status
    while ce_trade_status != 'completed' or pe_trade_status != 'completed':
        try:
            batch_update_values = [
                {"range": "A2", "values": [[selectedCEStrike]]},
                {"range": "B2", "values": [[ce_sell_time]]},
                {"range": "C2", "values": [[ce_sell_price]]},
                {"range": "D2", "values": [[ce_sl_price]]},
                {"range": "E2", "values": [[ce_buy_time]]},
                {"range": "F2", "values": [[ce_buy_price]]},
                {"range": "G2", "values": [[ce_exit_reason]]},
                {"range": "H2", "values": [[ce_limit_price]]},
                {"range": "I2", "values": [[ltp[selectedCEStrike]]]},
                {"range": "A3", "values": [[selectedPEStrike]]},
                {"range": "B3", "values": [[pe_sell_time]]},
                {"range": "C3", "values": [[pe_sell_price]]},
                {"range": "D3", "values": [[pe_sl_price]]},
                {"range": "E3", "values": [[pe_buy_time]]},
                {"range": "F3", "values": [[pe_buy_price]]},
                {"range": "G3", "values": [[pe_exit_reason]]},
                {"range": "H3", "values": [[pe_limit_price]]},
                {"range": "I3", "values": [[ltp[selectedPEStrike]]]},
            ]
            sh.batch_update(batch_update_values, value_input_option="USER_ENTERED")
        except Exception as e:
            logger.error(e)
            sys.exit(1)
    time.sleep(3)

但触发了WriteRequestsPerMinutePerUser的429错误：

{'code': 429, 'message': "Quota exceeded for quota metric 'Write requests' and limit 'Write requests per minute per user' of service 'sheets.googleapis.com' for consumer 'project_number:XXXX'.", 'status': 'RESOURCE_EXHAUSTED', 'details': [{'@type': 'type.googleapis.com/google.rpc.ErrorInfo', 'reason': 'RATE_LIMIT_EXCEEDED', 'domain': 'googleapis.com', 'metadata': {'service': 'sheets.googleapis.com', 'quota_location': 'global', 'quota_limit': 'WriteRequestsPerMinutePerUser', 'consumer': 'projects/XXXX', 'quota_metric': 'sheets.googleapis.com/write_requests', 'quota_limit_value': '120'}}, {'@type': 'type.googleapis.com/google.rpc.Help', 'links': [{'description': 'Request a higher quota limit.', 'url': 'https://cloud.google.com/docs/quota#requesting_higher_quota'}]}]}

当前WriteRequestsPerMinutePerUser配额为每分钟120次，使用batch_update本应整组更新算单次请求，设置3秒延迟后仍超配额，已提交配额提升申请，需其他解决办法。

核心原因 #

你的代码中time.sleep(3)放在while循环外部，只有当ce_trade_status和pe_trade_status都变为completed后才会执行一次sleep。这意味着，只要交易状态未完成，循环会无延迟地反复发送batch_update请求，短时间内请求量会远超每分钟120次的配额上限，直接触发429错误。

修复方案 #

1. 调整sleep位置到循环内部 #

将time.sleep(3)移到while循环的每次迭代中，确保每次请求后都等待3秒，这样每分钟最多发送20次请求，远低于120次的配额：

def update_gsheet():
    global ce_trade_status, pe_trade_status
    while ce_trade_status != 'completed' or pe_trade_status != 'completed':
        try:
            batch_update_values = [
                {"range": "A2", "values": [[selectedCEStrike]]},
                {"range": "B2", "values": [[ce_sell_time]]},
                # ... 省略其他项
                {"range": "I3", "values": [[ltp[selectedPEStrike]]]},
            ]
            sh.batch_update(batch_update_values, value_input_option="USER_ENTERED")
            # 每次请求后等待3秒
            time.sleep(3)
        except Exception as e:
            logger.error(e)
            # 遇到错误不要直接退出，可添加重试逻辑
            time.sleep(5)

2. 优化batch_update请求结构 #

当前的batch_update拆分成了18个独立range，虽然仍算1次请求，但可以合并连续单元格减少操作数，让请求更高效：

# 合并A2:I2和A3:I3的连续范围
batch_update_values = [
    {"range": "A2:I2", "values": [[
        selectedCEStrike, ce_sell_time, ce_sell_price, ce_sl_price,
        ce_buy_time, ce_buy_price, ce_exit_reason, ce_limit_price, ltp[selectedCEStrike]
    ]]},
    {"range": "A3:I3", "values": [[
        selectedPEStrike, pe_sell_time, pe_sell_price, pe_sl_price,
        pe_buy_time, pe_buy_price, pe_exit_reason, pe_limit_price, ltp[selectedPEStrike]
    ]]}
]

3. 添加429错误重试机制 #

遇到配额超限错误时，不要直接退出，而是采用指数退避策略重试，避免浪费配额：

from googleapiclient.errors import HttpError
import time

def update_gsheet():
    global ce_trade_status, pe_trade_status
    retry_delay = 3  # 初始延迟3秒
    max_retries = 5
    while ce_trade_status != 'completed' or pe_trade_status != 'completed':
        try:
            # 构建batch_update请求...
            sh.batch_update(batch_update_values, value_input_option="USER_ENTERED")
            retry_delay = 3  # 请求成功后重置延迟
            time.sleep(retry_delay)
        except HttpError as e:
            if e.resp.status == 429 and max_retries > 0:
                logger.warning(f"配额超限，{retry_delay}秒后重试")
                time.sleep(retry_delay)
                retry_delay *= 2  # 指数退避
                max_retries -= 1
            else:
                logger.error(e)
                break
        except Exception as e:
            logger.error(e)
            time.sleep(5)

4. 仅在数据变化时发送请求 #

维护一份上次更新的缓存数据，只有当数据实际发生变化时才调用batch_update，避免无意义的重复请求：

# 初始化缓存
last_ce_data = None
last_pe_data = None

def update_gsheet():
    global ce_trade_status, pe_trade_status, last_ce_data, last_pe_data
    while ce_trade_status != 'completed' or pe_trade_status != 'completed':
        # 收集当前数据
        current_ce_data = [
            selectedCEStrike, ce_sell_time, ce_sell_price, ce_sl_price,
            ce_buy_time, ce_buy_price, ce_exit_reason, ce_limit_price, ltp[selectedCEStrike]
        ]
        current_pe_data = [
            selectedPEStrike, pe_sell_time, pe_sell_price, pe_sl_price,
            pe_buy_time, pe_buy_price, pe_exit_reason, pe_limit_price, ltp[selectedPEStrike]
        ]
        
        # 仅当数据变化时更新
        if current_ce_data != last_ce_data or current_pe_data != last_pe_data:
            try:
                batch_update_values = [
                    {"range": "A2:I2", "values": [current_ce_data]},
                    {"range": "A3:I3", "values": [current_pe_data]}
                ]
                sh.batch_update(batch_update_values, value_input_option="USER_ENTERED")
                # 更新缓存
                last_ce_data = current_ce_data.copy()
                last_pe_data = current_pe_data.copy()
            except Exception as e:
                logger.error(e)
        
        time.sleep(3)

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