You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

基于object.finalized触发器的GCS转Google Drive代码报错求助

avatar
云原生阿岚
2026-6-27
问题：GCS文件复制到Google Drive的Cloud Function报错AttributeError

问题背景

通过object.finalized类型Cloud Function触发器，将GCS存储桶文件复制到指定Google Drive文件夹，代码运行后出现AttributeError: 'str' object has no attribute 'write'错误。

原代码 

import functions_framework
import base64
import tempfile
import traceback
from google.cloud import storage
from googleapiclient.discovery import build
import google.auth
from googleapiclient.http import MediaFileUpload

def upload_to_drive(temp_file, remote_path, credentials):
    """Uploads a file to the specified folder in Google Drive.

    Args:
        temp_file (tempfile.NamedTemporaryFile): Temporary file object containing the data to upload.
        remote_path (str): Name of the file to be created in Google Drive.
        credentials (google.auth.credentials.Credentials): Authentication credentials.
    """
    folder_id = str("1nNqtRloaFT0f71").split(",")
    service = build('drive', 'v3', credentials=credentials)
    print(f"Uploading file: {temp_file.name} to Google Drive folder: {folder_id}")

    # Use media body directly for efficient upload from temporary file
    response = service.files().create(
        body={
            "name": remote_path.split("/")[-1],
            "parents": folder_id,
            "mimeType": "text/csv"  # Set MIME type for CSV files
        },
        media_body=temp_file
    ).execute()
    print(f"File uploaded successfully: {response['id']}")

def download_file_from_gcs(bucket, key):
  """
  Downloads a file from Cloud Storage and writes it to a temporary file.

  Args:
      bucket (str): Name of the Cloud Storage bucket.
      key (str): Name of the file in the bucket.

  Returns:
      str: Path to the downloaded file within the temporary location.
  """
  print(f"Downloading key: {key} in bucket: {bucket}")
  client = storage.Client()
  source_bucket = client.bucket(bucket)
  blob_object = source_bucket.blob(key)
  with tempfile.NamedTemporaryFile(delete=False) as temp_file:
      blob_object.download_to_file(temp_file.name)
      local_path = temp_file.name  # Capture the temporary file path
  return local_path

# Triggered by a change in a storage bucket
@functions_framework.cloud_event
def drive_upload(cloud_event):
    data = cloud_event.data
    print("Full event data:", data)
    try:
        # Access bucket name assuming a nested structure
        bucket_name = data["bucket"]
        file_name = data["name"]
        print(f"File name extracted from event data: {file_name}")
        print(f"Bucket name extracted from event data: {bucket_name}")
    except KeyError:
        print("Bucket or file name not found in event data.")
        return  # Gracefully exit the function

    if file_name in ["Trials", "Control"]:
        try:
            # Credentials for Google Drive API access
            credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/drive"])
            # Downloading file from GCS with temporary file
            with tempfile.NamedTemporaryFile(delete=False) as temp_file:
                local = download_file_from_gcs(bucket_name, file_name)
                print(f("Temp File Path: {local}"))
                # Uploading the file to drive
                upload_to_drive(temp_file, file_name, credentials)
                print(f'File {file_name} copied to Google Drive folder')
        except Exception as e:
            print(traceback.format_exc())

错误日志 

2024-03-17 23:18:25.996 SAST Traceback (most recent call last):   File "/workspace/main.py", line 74, in drive_upload     local = download_file_from_gcs(bucket_name, file_name)
2024-03-17 23:18:25.996 SAST             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

2024-03-17 23:18:25.996 SAST File "/workspace/main.py", line 49, in download_file_from_gcs

2024-03-17 23:18:25.996 SAST blob_object.download_to_file(temp_file.name)

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 1150, in download_to_file

2024-03-17 23:18:25.996 SAST self._prep_and_do_download(

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 4352, in _prep_and_do_download

2024-03-17 23:18:25.996 SAST self._do_download(

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 1007, in _do_download

2024-03-17 23:18:25.996 SAST response = download.consume(transport, timeout=timeout)
2024-03-17 23:18:25.996 SAST            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/resumable_media/requests/download.py", line 237, in consume

2024-03-17 23:18:25.996 SAST return _request_helpers.wait_and_retry(
2024-03-17 23:18:25.996 SAST        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/resumable_media/requests/_request_helpers.py", line 155, in wait_and_retry

2024-03-17 23:18:25.996 SAST response = func()
2024-03-17 23:18:25.996 SAST            ^^^^^^

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/resumable_media/requests/download.py", line 233, in retriable_request

2024-03-17 23:18:25.996 SAST self._write_to_stream(result)

2024-03-17 23:18:25.996 SAST File "/layers/google.python.pip/pip/lib/python3.12/site-packages/google/resumable_media/requests/download.py", line 124, in _write_to_stream

2024-03-17 23:18:25.996 SAST  self._stream.write(chunk)
2024-03-17 23:18:25.996 SAST  ^^^^^^^^^^^^^^^^^^

2024-03-17 23:18:25.996 SAST  AttributeError: 'str' object has no attribute 'write'

问题分析

  1. 核心错误原因：download_file_from_gcs函数中，blob_object.download_to_file方法需要接收打开的文件对象（具备write方法），但传入的是temp_file.name（字符串类型的文件路径），导致内部调用write时触发AttributeError。
  2. 次要问题：
    • drive_upload中重复创建临时文件，download_file_from_gcs内部已创建临时文件，外层无需再创建
    • print(f("Temp File Path: {local}"))存在语法错误，多了一层括号
    • folder_id = str("1nNqtRloaFT0f71").split(",")写法冗余，直接定义列表即可
    • upload_to_drive接收的临时文件对象与实际下载的文件路径不匹配，逻辑混乱

修复方案

修正后的完整代码 

import functions_framework
import tempfile
import traceback
from google.cloud import storage
from googleapiclient.discovery import build
import google.auth
from googleapiclient.http import MediaFileUpload

def upload_to_drive(file_path, remote_path, credentials):
    """Uploads a file to the specified folder in Google Drive.

    Args:
        file_path (str): Local path to the file to upload.
        remote_path (str): Name of the file to be created in Google Drive.
        credentials (google.auth.credentials.Credentials): Authentication credentials.
    """
    # 直接定义目标文件夹ID列表
    folder_id = ["1nNqtRloaFT0f71"]
    service = build('drive', 'v3', credentials=credentials)
    print(f"Uploading file: {file_path} to Google Drive folder: {folder_id}")

    # 使用MediaFileUpload加载本地文件
    media_body = MediaFileUpload(file_path, mimetype="text/csv")
    response = service.files().create(
        body={
            "name": remote_path.split("/")[-1],
            "parents": folder_id
        },
        media_body=media_body
    ).execute()
    print(f"File uploaded successfully: {response['id']}")

def download_file_from_gcs(bucket, key):
    """
    Downloads a file from Cloud Storage and writes it to a temporary file.

    Args:
        bucket (str): Name of the Cloud Storage bucket.
        key (str): Name of the file in the bucket.

    Returns:
        str: Path to the downloaded file within the temporary location.
    """
    print(f"Downloading key: {key} in bucket: {bucket}")
    client = storage.Client()
    source_bucket = client.bucket(bucket)
    blob_object = source_bucket.blob(key)
    
    # 使用download_to_filename直接下载到临时文件路径，无需手动打开文件对象
    with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as temp_file:
        blob_object.download_to_filename(temp_file.name)
        local_path = temp_file.name
    
    return local_path

# Triggered by a change in a storage bucket
@functions_framework.cloud_event
def drive_upload(cloud_event):
    data = cloud_event.data
    print("Full event data:", data)
    try:
        bucket_name = data["bucket"]
        file_name = data["name"]
        print(f"File name extracted from event data: {file_name}")
        print(f"Bucket name extracted from event data: {bucket_name}")
    except KeyError:
        print("Bucket or file name not found in event data.")
        return

    if file_name in ["Trials", "Control"]:
        try:
            # 获取Drive API授权凭证
            credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/drive"])
            # 从GCS下载文件到本地临时路径
            local_file_path = download_file_from_gcs(bucket_name, file_name)
            print(f"Temp File Path: {local_file_path}")
            # 上传到Google Drive
            upload_to_drive(local_file_path, file_name, credentials)
            print(f'File {file_name} copied to Google Drive folder')
        except Exception as e:
            print(traceback.format_exc())

关键修复点

  1. 将download_file_from_gcs中的download_to_file(temp_file.name)改为download_to_filename(temp_file.name)，该方法直接接收文件路径，无需传入文件对象。
  2. 重构upload_to_drive，改为接收文件路径，使用MediaFileUpload加载文件，避免文件对象关闭导致的问题。
  3. 移除drive_upload中多余的临时文件创建逻辑，直接使用download_file_from_gcs返回的路径。
  4. 修复print语句的语法错误，去掉多余括号。
  5. 简化folder_id的定义方式，直接用列表存储目标文件夹ID。

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.06.27 22:48:09