仅6次请求触发Google Drive API用户速率限制问题求助

问题描述 #

上传文件至Google Drive共享驱动器时，触发以下403错误：

Error: <HttpError 403 when requesting None returned "User rate limit exceeded". Details: "[{'message': 'User rate limit exceeded', 'domain': 'usageLimits', 'reason': 'userRateLimitExceeded', 'location': 'user', 'locationType': 'other'}]">

查看配额页面显示仅发起6次请求，远低于每分钟12000次的全局配额；相同代码在个人驱动器中可正常上传，已添加退避逻辑但问题依旧，当前请求间隔约350ms（每秒3次）。

相关代码：

import time
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload

def upload_file(drive_service, folder_id, filename):
    delay = 0  # start with no backoff

    while True:  # keep trying until the file is successfully uploaded
        try:
            time.sleep(delay)  # backoff

            # Create a file in the Drive with a numbered name
            file_metadata = {
                'name': filename,
                'parents': [folder_id]
            }
            media = MediaFileUpload(filename, mimetype='image/jpeg')
            file = drive_service.files().create(body=file_metadata, media_body=media, fields='id').execute()

            # If the upload was successful, break the loop
            break
        except Exception as e:
            if '403' in str(e):  # if rejected with a rate limit
                delay += 2  # add 2s to delay to guarantee the next attempt will be OK
            print(f"Error: {e}, retrying in {delay} seconds")

        if delay > 0:  # if we are currently backing off
            delay = max(0, delay - 0.2)  # reduce the backoff delay

def main():
    # Load the credentials from 'client_secret.json'
    flow = InstalledAppFlow.from_client_secrets_file(
        '[Path to your client secret json]',
        scopes=['https://www.googleapis.com/auth/drive']
    )
    creds = flow.run_local_server(port=8080)

    # Build the Drive service
    drive_service = build('drive', 'v3', credentials=creds)

    # Specify the ID of the folder in Google Drive where the images will be uploaded
    folder_id = '[Google Drive folder ID]'

    # Specify the filename of the image to be uploaded
    filename = 'image.jpg'

    # Upload the file
    upload_file(drive_service, folder_id, filename)

if __name__ == "__main__":
    main()

解决方案 #

1. 理解共享驱动器的特殊配额规则 #

共享驱动器的速率限制与个人驱动器不通用，即使单用户请求量远低于全局配额，也可能因共享驱动器的组织级限制、并发操作阈值或驱动器整体负载触发拦截，这类限制不会直接显示在普通配额页面。

2. 修正退避逻辑（关键） #

当前退避逻辑存在缺陷：错误后固定加2秒但后续会递减，无法满足共享驱动器的严格要求。需改用指数退避+随机抖动，并优先遵循API返回的Retry-After响应头：

import random

def upload_file(drive_service, folder_id, filename):
    delay = 1  # 初始退避1秒
    max_delay = 64  # 最大退避上限，避免无限等待
    while True:
        try:
            time.sleep(delay)
            file_metadata = {
                'name': filename,
                'parents': [folder_id]
            }
            media = MediaFileUpload(filename, mimetype='image/jpeg')
            file = drive_service.files().create(body=file_metadata, media_body=media, fields='id').execute()
            print(f"文件上传成功，ID: {file.get('id')}")
            break
        except Exception as e:
            error_str = str(e)
            if '403' in error_str:
                # 优先使用API返回的重试时间
                if hasattr(e, 'resp') and 'Retry-After' in e.resp.headers:
                    delay = int(e.resp.headers['Retry-After'])
                else:
                    # 指数退避+随机抖动，避免并发请求冲突
                    delay = min(delay * 2 + random.uniform(0, 1), max_delay)
                print(f"速率限制触发，{delay}秒后重试")
            else:
                # 非速率限制错误直接抛出，避免无限循环
                print(f"上传失败: {e}")
                raise

3. 检查组织级配额限制 #

若账号属于Google Workspace组织，管理员可能设置了自定义的共享驱动器操作配额，需联系管理员确认是否存在更严格的速率限制政策。

4. 排查多来源请求 #

确认无其他脚本/应用使用同一账号向该共享驱动器发起请求，累计请求量可能触发限制。可在Google Cloud Console的API仪表板查看详细请求日志，核对所有来源的请求总量。

5. 批量操作优化（多文件场景） #

若需上传多个文件，改用批量请求减少HTTP请求次数，降低触发限制的概率：

from googleapiclient.http import BatchHttpRequest

def batch_upload(drive_service, folder_id, filenames):
    def callback(request_id, response, exception):
        if exception:
            print(f"上传失败: {exception}")
        else:
            print(f"文件上传成功，ID: {response.get('id')}")

    batch = drive_service.new_batch_http_request(callback=callback)
    for filename in filenames:
        file_metadata = {'name': filename, 'parents': [folder_id]}
        media = MediaFileUpload(filename, mimetype='image/jpeg')
        batch.add(drive_service.files().create(body=file_metadata, media_body=media, fields='id'))
    
    batch.execute()

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