Google Cloud Function上传Drive文件至GCS时文件名获取报错

你遇到的报错和功能异常来自4个核心代码逻辑错误：

• 调用Drive API的路径错误：Drive v3 SDK的根服务对象没有直接的get()方法，单文件查询接口的完整路径是service.files().get()，漏写.files()层级会直接触发你看到的属性错误。
• 文件列表遍历逻辑错误：service.files().list().execute()返回的是顶层字典，文件列表存在字典的files键下，直接遍历顶层字典只会拿到kind、files这类元数据键名，拿不到具体文件对象，代码里未定义的fileId变量也会触发后续异常。
• 上传逻辑错误：blob.upload_from_filename()仅支持上传运行环境本地磁盘的文件，Cloud Function本地没有Google Drive里的文件副本，直接传文件名会报文件不存在错误，必须先把Drive文件内容拉取到内存流再上传。
• 冗余请求：你在调用list()接口时已经在fields参数里声明要返回files(id,name)，不需要再单独发一次get()请求拿文件名，平白浪费API配额。

首先清理掉Cloud Function环境下不需要的冗余导入（比如桌面端OAuth授权相关的InstalledAppFlow、未使用的pandas依赖等），修正API调用路径和上传逻辑，完整可运行代码如下：

import io
import logging
import google.auth
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload
from google.cloud import storage

# 初始化API客户端
SCOPES = ['https://www.googleapis.com/auth/drive']
creds, project = google.auth.default(scopes=SCOPES)
drive_service = build('drive', 'v3', credentials=creds)
storage_client = storage.Client()

# 配置项
BUCKET_NAME = 'my-bucket'
TEAM_DRIVE_FOLDER_ID = 'xxxxxxxxxxxxxxxxxA'
GCS_TARGET_PREFIX = "incoming/iri/IRI_Updates/Ongoing_Sales_Data/2022/"
QUERY = "name contains 'Customer' and name contains '2022' and trashed = false"

def iri_data_sync(data, context):
    # 拉取符合条件的Drive文件列表
    resp = drive_service.files().list(
        q=QUERY,
        driveId=TEAM_DRIVE_FOLDER_ID,
        supportsAllDrives=True,
        includeItemsFromAllDrives=True,
        corpora='drive',
        fields="files(id,name)"
    ).execute()
    file_list = resp.get('files', [])
    if not file_list:
        logging.info("No matched files found in Drive.")
        return

    bucket = storage_client.bucket(BUCKET_NAME)
    for file in file_list:
        file_id = file['id']
        file_name = file['name']
        logging.info(f"Start processing file: {file_name} (ID: {file_id})")

        # 把Drive文件内容下载到内存流，不落本地盘
        request = drive_service.files().get_media(fileId=file_id)
        file_stream = io.BytesIO()
        downloader = MediaIoBaseDownload(file_stream, request)
        done = False
        while not done:
            status, done = downloader.next_chunk()
            logging.info(f"Download {int(status.progress() * 100)}%")
        file_stream.seek(0)

        # 上传流到GCS
        blob = bucket.blob(f"{GCS_TARGET_PREFIX}{file_name}")
        blob.upload_from_file(file_stream)
        logging.info(f"Successfully uploaded {file_name} to GCS.")

• 部署Cloud Function时，要给绑定的服务账号授予两个权限：目标共享盘的查看者权限、目标GCS存储桶的存储对象创建者权限，否则会报权限拒绝错误。
• 如果符合条件的文件超过1000个，需要在list接口里处理pageToken分页参数，否则只能拉取第一页的文件。
• 大文件场景下建议给Cloud Function分配足够的内存，避免内存溢出，单文件超过100MB时建议改用分片上传逻辑。

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