GCP云函数部署Python脚本调用Gmail API与GCS的鉴权配置方案

Cloud Function运行时内置了默认服务账号的身份凭证，所有GCP服务的客户端都会自动拾取这个凭证做鉴权，不需要手动上传、读取本地JSON密钥文件，只需要做好权限配置和少量代码适配即可。

第一步：权限配置 #

• GCS存储权限：找到Cloud Function绑定的运行时服务账号（默认格式为PROJECT_ID@appspot.gserviceaccount.com），给该账号分配目标存储桶的存储对象创建者权限，即可直接调用GCS接口，不需要额外密钥。
• Gmail API权限：
  • 如果使用Google Workspace企业邮箱：在GCP控制台启用Gmail API后，到Workspace管理后台给该服务账号开启域级授权，添加https://mail.google.com/作用域，服务账号即可直接模拟目标邮箱身份调用接口，完全不需要本地的credentials.json文件。
  • 如果使用个人Gmail账号：域级授权不可用，可以把原来credentials.json文件的全部内容复制为Cloud Function的环境变量（比如变量名设为GMAIL_OAUTH_CRED），初始化Gmail服务时直接从环境变量读取凭证内容即可，不需要把json文件打包进部署包。注意首次授权需要提前在本地完成OAuth授权流程，把生成的refresh_token也存到环境变量里，避免云函数环境无法弹出浏览器完成授权。

第二步：代码修改点 #

1. 删掉手动设置os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = r'storage_account.json'这行代码，GCS客户端初始化时会自动加载云函数内置的服务账号凭证。
2. 删掉本地调试用的if __name__ == '__main__'代码块，把Gmail服务的初始化逻辑移到全局，不要依赖读取本地credentials.json文件。
3. 临时文件存储路径改到云函数唯一可写的/tmp目录，写完上传GCS后及时清理临时文件，避免磁盘空间不足。
4. 添加符合Cloud Function规范的入口函数，部署时指定该函数为触发入口。
5. 原来依赖的自定义create_service方法可以替换为官方SDK的默认凭证构建逻辑，不需要额外封装读本地文件的逻辑。

第三步：部署用最终代码 #

import os
import base64
from typing import List
import time
import json
from googleapiclient.discovery import build
from google.auth import default
from google.oauth2.credentials import Credentials
from google.cloud import storage

class GmailException(Exception):
    """Gmail基础异常类"""

class NoEmailFound(GmailException):
    """未查询到匹配邮件"""

# 初始化GCS客户端，自动拾取云函数内置服务账号凭证
storage_client = storage.Client()
bucket_name = 'mybucketname'

# 初始化Gmail API服务
SCOPES = ['https://mail.google.com/']
# 优先使用云函数服务账号默认凭证（Workspace域委派场景）
try:
    credentials, _ = default(scopes=SCOPES)
# 个人Gmail账号场景：从环境变量读取OAuth凭证
except:
    gmail_cred_data = json.loads(os.environ.get('GMAIL_OAUTH_CRED'))
    credentials = Credentials(
        token=gmail_cred_data.get('token'),
        refresh_token=gmail_cred_data.get('refresh_token'),
        token_uri=gmail_cred_data.get('token_uri', 'https://oauth2.googleapis.com/token'),
        client_id=gmail_cred_data.get('client_id'),
        client_secret=gmail_cred_data.get('client_secret'),
        scopes=SCOPES
    )
service = build('gmail', 'v1', credentials=credentials)

def search_emails(query_string: str, label_ids: List=None):
    try:
        message_list_response = service.users().messages().list(
            userId='me',
            labelIds=label_ids,
            q=query_string
        ).execute()

        message_items = message_list_response.get('messages', [])
        next_page_token = message_list_response.get('nextPageToken')
        
        while next_page_token:
            message_list_response = service.users().messages().list(
                userId='me',
                labelIds=label_ids,
                q=query_string,
                pageToken=next_page_token
            ).execute()

            message_items.extend(message_list_response.get('messages', []))
            next_page_token = message_list_response.get('nextPageToken')
        return message_items
    except Exception as e:
        raise NoEmailFound('No emails returned')

def get_file_data(message_id, attachment_id, file_name, new_Location):
    response = service.users().messages().attachments().get(
        userId='me',
        messageId=message_id,
        id=attachment_id
    ).execute()
    
    file_data = base64.urlsafe_b64decode(response.get('data').encode('UTF-8'))
    return file_data

def get_message_detail(message_id, msg_format='metadata', metadata_headers: List=None):
    message_detail = service.users().messages().get(
        userId='me',
        id=message_id,
        format=msg_format,
        metadataHeaders=metadata_headers
    ).execute()
    return message_detail

def upload_to_bucket(blob_name, file_path, bucket_name):
    try:
        bucket = storage_client.get_bucket(bucket_name)
        blob = bucket.blob(blob_name)
        blob.upload_from_filename(file_path)
        return True
    except Exception as e:
        print(e)
        return False

def save_file_data(email_messages):
    for email_message in email_messages: 
        messageDetail = get_message_detail(email_message['id'], msg_format='full', metadata_headers=['parts'])  
        headers=messageDetail["payload"]["headers"]
        subjects= [j['value'] for j in headers if j["name"]=="Subject"]
        subject=subjects[0]      
        save_location=subject
        messageDetailPayload = messageDetail.get('payload') 
        if 'parts' in messageDetailPayload: 
            for msgPayload in messageDetailPayload['parts']: 
                file_name = msgPayload['filename'] 
                filetype = ".csv"
                if file_name.find(filetype) != -1:
                    body = msgPayload['body'] 
                    if 'attachmentId' in body: 
                        attachment_id = body['attachmentId'] 
                        attachment_content = get_file_data(email_message['id'], attachment_id, file_name, save_location)
                        # 临时文件存到云函数可写的/tmp目录
                        temp_file_path = os.path.join('/tmp', file_name)
                        with open(temp_file_path, "wb") as binary_file:
                            binary_file.write(attachment_content)
                            upload_to_bucket(f'{save_location}/{file_name}', temp_file_path, bucket_name)
                            # 上传完成删除临时文件
                            os.remove(temp_file_path)
                            print(f'File {file_name} is saved at gs://{bucket_name}/{save_location}/') 
        time.sleep(0.5)

# 云函数入口，支持HTTP、Pub/Sub等常见触发方式
def main(event, context):
    query_string = 'has:attachment' 
    email_messages = search_emails(query_string) 
    save_file_data(email_messages)
    return "Processing completed"

部署注意事项 #

• 部署时requirements.txt需要包含以下依赖：

  google-cloud-storage
  google-api-python-client
  google-auth
  

• 不需要把本地的storage_account.json、credentials.json打包进部署包。
• 部署时入口函数填写main即可。

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