Gmail API批量拉取邮件时随机出现HttpError 400前置条件失败问题

错误现象 #

去年编写的代码原本可正常拉取前一晚到次日清晨的邮件，但近期批量拉取过程中随机出现如下错误：

Error: <HttpError 400 when requesting https://gmail.googleapis.com/gmail/v1/users/me/messages/blah?format=full&alt=json returned "Precondition check failed.". Details: "[{'message': 'Precondition check failed.', 'domain': 'global', 'reason': 'failedPrecondition'}]">, cannot fetch message

直接点击该URL返回401未授权（缺少OAuth2凭证）：

{
  "error": {
    "code": 401,
    "message": "Request is missing required authentication credential. Expected OAuth 2 access token, login cookie or other valid authentication credential.",
    "errors": [
      {
        "message": "Login Required.",
        "domain": "global",
        "reason": "required",
        "location": "Authorization",
        "locationType": "header"
      }
    ],
    "status": "UNAUTHENTICATED",
    "details": [
      {
        "@type": "type.googleapis.com/google.rpc.ErrorInfo",
        "reason": "CREDENTIALS_MISSING",
        "domain": "googleapis.com",
        "metadata": {
          "method": "caribou.api.proto.MailboxService.GetMessage",
          "service": "gmail.googleapis.com"
        }
      }
    ]
  }
}

奇怪的是代码中途能拉取部分邮件，错误完全随机：有时10次仅1次成功，有时1小时内100%正常运行。

代码实现 #

from __future__ import print_function
import sqlite3
from datetime import datetime, timedelta
import pytz
import base64
import logging
import time
import pendulum
import os.path

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

from google_auth_oauthlib.flow import InstalledAppFlow

def get_decode_body(payload):
    if "parts" not in payload:
        return ""
    email_content = ""
    for part in payload["parts"]:
        if part["mimeType"] == "text/plain":
            data = part["body"]["data"]
            email_content += base64.urlsafe_b64decode(data).decode("utf-8")
        else:
            email_content += get_decode_body(part)
    return email_content

def get_msg_date(message):
    return datetime.fromtimestamp(float(message["internalDate"]) / 1000.0).astimezone(pytz.timezone("US/Eastern"))

def check_date_boundary(message, str_date, end_date):
    dt = get_msg_date(message)
    return str_date <= dt and end_date >= dt

def get_subject(message):
    headers = message["payload"]["headers"]
    # checks for empty subject
    subject_list = [i['value'] for i in headers if i["name"]=="Subject"]
    subject = "".join(subject_list)
    return subject

def convert_to_datetime(time_string):
    TIME_FORMAT = "%y%m%d %H:%M"
    return datetime.strptime(time_string, TIME_FORMAT).astimezone(pytz.timezone("US/Eastern"))

# used to just call Gmail API
class Gmail:

    def __init__(self, token_file):
        # The file token.json stores the user's access and refresh tokens, and is
        # created automatically when the authorization flow completes for the first
        # time
        creds = None
        if os.path.exists(token_file):
            creds = Credentials.from_authorized_user_file(token_file, ['https://mail.google.com/'])
        # If there are no (valid) credentials available, let the user log in.
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                print("LOGGER: token invalid, refreshing for new token")
                creds.refresh(Request())
            else:
                print("LOGGER: re-credentialing due to token not found")
                flow = InstalledAppFlow.from_client_secrets_file(
                    'credentials.json', ['https://mail.google.com/'])
                creds = flow.run_local_server(port=0)
            # Save the credentials for the next run
            with open('token.json', 'w') as token:
                token.write(creds.to_json())

        self.service = build('gmail', 'v1', credentials=creds, cache_discovery=False)

    def get_all_messages(self, start_dt, end_dt, page_token = None):
        # Gmail does not give results when before and after dates are the same, edate=edate+1 will be fine for 9 to 9:20 run
        # Padding is done on sdate/edate to account for utc time difference when calling Gmail API query
        sdate = start_dt.date()-timedelta(days=1)
        edate = end_dt.date()+timedelta(days=1)

        messages_out = []

        # loop is based on page_token, first and last page of search query is 0 but between that there is #
        while True:
            results = self.service.users().messages().list(userId='me', includeSpamTrash=True, q='before:%s, after:%s' % (edate, sdate), pageToken=page_token, maxResults=45).execute()

            message_ids = results.get('messages', [])
            messages = []
            def add(id, msg, err):
                # id is given because this will not be called in the same order
                if err:
                    raise Exception("Error: %s, cannot fetch message" % (err))
                else:
                    # print(f"LOGGER: append msg id {msg['id']} to list of messages")
                    messages.append(msg)

            page_token = results["nextPageToken"] if "nextPageToken" in results else None
            batch = self.service.new_batch_http_request()
            print("LOGGER: creating new batch http request")
            for i, msg in enumerate(message_ids):
                #print(f"LOGGER: getting message {i} by id {msg['id']} and adding to batch request")
                batch.add(self.service.users().messages().get(userId='me', id=msg['id'], format='full'), callback=add)
            print("LOGGER: executing batch request")
            batch.execute()
            time.sleep(1)
            # fetching batch of message based on current page_token

            if not messages:
                print('LOGGER: No messages found.')
            else:
                criteria = '(SINCE "' + start_dt.strftime("%d-%b-%Y") + '")'
                print(f"LOGGER: {str(datetime.now())} - Page at {criteria}")

                # getting subject, decoded body message, timestamp of message, and the message ID
                for message in messages:
                    if check_date_boundary(message, start_dt, end_dt):
                        try:
                            body_decoded = get_decode_body(message["payload"])
                        except Exception as e:
                            print("LOGGER: Error decoding message", message)
                            raise e
                        subject = get_subject(message)
                        dt = get_msg_date(message)
                        messages_out.append({"subject": subject, "message": body_decoded, "datetime": dt, "msg_id": message["id"]})

            if page_token is None:
                break
        # messages_out contain all message items from all the pages
        return messages_out

token_path = os.path.join("/home/user/", "token.json")
gmail = Gmail(token_path)
start_dt = convert_to_datetime("240117 16:00")
end_dt = convert_to_datetime("240118 9:00")
print(gmail.get_all_messages(start_dt, end_dt))

已尝试的解决措施 #

• 重新认证获取新token
• 手动过期token.json
• 双重检查API密钥
• 将maxResults降至50以下以避免每秒配额限制

当前使用的库版本 #

• google-api-python-client 2.114.0
• google-auth 2.16.1
• google-auth-httplib2 0.2.0
• google-auth-oauthlib 1.2.0

问题分析与修复方案 #

核心原因推测 #

1. 批量请求中凭证过期：初始化service时的访问令牌有效期通常为1小时，若批量拉取耗时较长，后续请求会因令牌过期触发failedPrecondition错误，且批量请求不会自动刷新凭证。
2. 日期查询格式不兼容：Gmail API要求日期参数为DD-MMM-YYYY或YYYY/MM/DD格式，代码直接传入date()对象可能导致随机格式解析异常。
3. 批量回调的异常处理不当：回调中直接抛出异常会中断整个批量任务，且未考虑重试场景。

具体修复步骤 #

1. 每次批量请求前校验并刷新凭证

在执行批量请求前检查凭证有效性，过期则刷新并重建service：

# 在get_all_messages方法的batch.execute()前添加
if not self.service._credentials.valid:
    if self.service._credentials.expired and self.service._credentials.refresh_token:
        print("LOGGER: Refreshing expired token before batch request")
        self.service._credentials.refresh(Request())
        self.service = build('gmail', 'v1', credentials=self.service._credentials, cache_discovery=False)

2. 修正日期查询格式

将sdate和edate转换为Gmail API支持的DD-MMM-YYYY格式：

# 替换原sdate和edate的定义
sdate = (start_dt.date() - timedelta(days=1)).strftime("%d-%b-%Y")
edate = (end_dt.date() + timedelta(days=1)).strftime("%d-%b-%Y")

3. 改进批量请求的错误处理

收集错误而非直接抛出，避免单次请求失败中断整个任务：

# 修改get_all_messages中的messages和回调逻辑
messages = []
errors = []
def add(id, msg, err):
    if err:
        errors.append((id, err))
        print(f"LOGGER: Failed to fetch message {id}: {err}")
    else:
        messages.append(msg)

# 执行批量后处理错误
batch.execute()
if errors:
    print(f"LOGGER: {len(errors)} messages failed to fetch")
    # 可添加针对失败消息的重试逻辑

4. 添加错误重试机制

针对failedPrecondition和UNAUTHENTICATED错误添加自动重试：

# 自定义重试装饰器
def retry_on_failure(max_attempts=3):
    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if "failedPrecondition" in str(e) or "UNAUTHENTICATED" in str(e):
                        attempts += 1
                        print(f"LOGGER: Retrying attempt {attempts}/{max_attempts} due to error: {e}")
                        time.sleep(2 ** attempts)
                    else:
                        raise e
            raise Exception(f"LOGGER: Failed after {max_attempts} attempts")
        return wrapper
    return decorator

# 给get_all_messages方法添加重试
@retry_on_failure(max_attempts=3)
def get_all_messages(self, start_dt, end_dt, page_token=None):
    # 原方法逻辑

5. 更新依赖库

升级google-auth相关库，修复潜在的凭证刷新bug：

pip install --upgrade google-auth google-auth-oauthlib google-api-python-client

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