如何用Python覆盖Google Cloud Storage存储桶中的单个文件

我在Compute Engine虚拟机的某个位置有一个logs.txt文件，希望定期将其备份（覆盖）到Google Cloud Storage存储桶中。因为logs.txt是Python脚本预处理的结果，所以需要在脚本里完成上传操作（不考虑用gsutil cp命令）。Compute Engine实例和Cloud Storage存储桶属于同一个GCP项目，彼此可访问。

我目前基于官方示例写的代码如下：

from google.cloud import storage

bucket_name = "my-bucket"
destination_blob_name = "logs.txt"
source_file_name = "logs.txt"  # 脚本可访问该文件

storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(destination_blob_name)

generation_match_precondition = 0
blob.upload_from_filename(source_file_name, if_generation_match=generation_match_precondition)

print(f"File {source_file_name} uploaded to {destination_blob_name}.")

当gs://my-bucket/logs.txt不存在时，脚本运行正常，但尝试覆盖时会出现如下错误：

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 2571, in upload_from_file
    created_json = self._do_upload(
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 2372, in _do_upload
    response = self._do_multipart_upload(
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 1907, in _do_multipart_upload
    response = upload.transmit(
  File "/usr/local/lib/python3.8/dist-packages/google/resumable_media/requests/upload.py", line 153, in transmit
    return _request_helpers.wait_and_retry(
  File "/usr/local/lib/python3.8/dist-packages/google/resumable_media/requests/_request_helpers.py", line 147, in wait_and_retry
    response = func()
  File "/usr/local/lib/python3.8/dist-packages/google/resumable_media/requests/upload.py", line 149, in retriable_request
    self._process_response(result)
  File "/usr/local/lib/python3.8/dist-packages/google/resumable_media/_upload.py", line 114, in _process_response
    _helpers.require_status_code(response, (http.client.OK,), self._get_status_code)
  File "/usr/local/lib/python3.8/dist-packages/google/resumable_media/_helpers.py", line 105, in require_status_code
    raise common.InvalidResponse(
google.resumable_media.common.InvalidResponse: ('Request failed with status code', 412, 'Expected one of', <HTTPStatus.OK: 200>)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/my_folder/upload_to_gcs.py", line 76, in <module>
    blob.upload_from_filename(source_file_name, if_generation_match=generation_match_precondition)
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 2712, in upload_from_filename
    self.upload_from_file(
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 2588, in upload_from_file
    _raise_from_invalid_response(exc)
  File "/usr/local/lib/python3.8/dist-packages/google/cloud/storage/blob.py", line 4455, in _raise_from_invalid_response
    raise exceptions.from_http_status(response.status_code, message, response=response)
google.api_core.exceptions.PreconditionFailed: 412 POST https://storage.googleapis.com/upload/storage/v1/b/production-onementor-dt-data/o?uploadType=multipart&ifGenerationMatch=0: {
  "error": {
    "code": 412,
    "message": "At least one of the pre-conditions you specified did not hold.",
    "errors": [
      {
        "message": "At least one of the pre-conditions you specified did not hold.",
        "domain": "global",
        "reason": "conditionNotMet",
        "locationType": "header",
        "location": "If-Match"
      }
    ]
  }
}
: ('Request failed with status code', 412, 'Expected one of', <HTTPStatus.OK: 200>)

我查了upload_from_filename的文档，但没找到“启用覆盖”的标识。请问如何通过Python正确覆盖Google Cloud Storage存储桶中已存在的文件？

解决方案 #

问题出在你设置的generation_match_precondition = 0，这个参数的作用是仅当目标文件不存在（生成号为0）时才允许上传。当文件已经存在时，它的生成号大于0，就会触发412预条件失败错误。

要实现覆盖已有文件，有两种可行方法：

方法1：移除预条件参数 #

直接去掉if_generation_match参数，GCS默认允许覆盖已有文件：

from google.cloud import storage

bucket_name = "my-bucket"
destination_blob_name = "logs.txt"
source_file_name = "logs.txt"

storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(destination_blob_name)

# 直接上传，覆盖已有文件
blob.upload_from_filename(source_file_name)

print(f"File {source_file_name} uploaded to {destination_blob_name}.")

方法2：忽略生成号检查（可选） #

如果你想明确指定强制覆盖，可以使用if_generation_not_match=None，表示忽略生成号验证逻辑：

from google.cloud import storage

bucket_name = "my-bucket"
destination_blob_name = "logs.txt"
source_file_name = "logs.txt"

storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(destination_blob_name)

# 忽略生成号检查，强制覆盖
blob.upload_from_filename(source_file_name, if_generation_not_match=None)

print(f"File {source_file_name} uploaded to {destination_blob_name}.")

补充说明 #

• GCS中每个对象的generation（生成号）是每次修改后递增的数值，用于实现乐观锁机制。
• if_generation_match=N：仅当目标对象的生成号等于N时才允许操作，适合避免并发冲突场景。
• 不设置任何预条件参数时，默认行为就是覆盖已有对象。

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