ShowDocs
搜索文章Ctrl K

外观

← 返回Google Cloud Storage列表
应用实践

GCS 实践：带 generation 条件的完整 Python CLI ​

配置与前置条件 ​

GCS 配置需要一个读者预先准备的私有实验桶、受限 ADC、项目标识与独立前缀。前置：前提条件。本轮示例未执行，没有安装依赖、连接账号或调用 API。Python 3.10+，客户端声明 3.x；这不是锁定的可重复构建清单，读者自行安装后应保存实际解析版本。

txt
# The declared 3.x API family; not installed or resolved during authoring.
google-cloud-storage>=3.0,<4
google-crc32c>=1.6,<2
ini
# Documentation only. Set variables in your shell; this file is not automatically loaded.
GOOGLE_CLOUD_PROJECT=replace-with-lab-project
GCS_BUCKET=replace-with-existing-private-lab-bucket
GCS_PREFIX=theory-lab/your-session/
# Optional ADC configuration file path; prefer federation, never commit the file.
GOOGLE_APPLICATION_CREDENTIALS=

环境示例仅作配置说明，不被 CLI 加载。若使用 GOOGLE_APPLICATION_CREDENTIALS，应填联合身份配置或已经获准的凭据文件路径，绝不提交文件内容；若使用附加身份，可不设此变量。不要设置一个空字符串后期待它等同于已验证的凭据路径。GOOGLE_CLOUD_PROJECT 并不替代 Bucket IAM。

同一份完整源码 ​

py
"""GCS 3.x CLI reading sample. No bucket or IAM creation."""
import argparse
import json
import os
import sys
import uuid
from pathlib import Path

from google.cloud import storage
from google.api_core.exceptions import GoogleAPICallError


def main():
    parser = argparse.ArgumentParser(description="Existing private GCS lab bucket only")
    parser.add_argument("command", choices=["list", "put", "get", "delete", "sign"])
    parser.add_argument("--key")
    parser.add_argument("--file")
    parser.add_argument("--generation", type=int)
    args = parser.parse_args()
    client = None
    try:
        prefix = os.environ["GCS_PREFIX"]
        if not prefix.startswith("theory-lab/") or not prefix.endswith("/"):
            raise ValueError("GCS_PREFIX must start theory-lab/ and end /")
        key = args.key
        if args.command == "put":
            if not args.file:
                raise ValueError("put requires --file")
            key = prefix + str(uuid.uuid4()) + ".bin"
        elif args.command != "list" and not key:
            raise ValueError("--key is required")
        if key and (not key.startswith(prefix) or ".." in key.split("/")):
            raise ValueError("key outside configured prefix")
        if args.command == "delete" and args.generation is None:
            raise ValueError("delete requires --generation")
        client = storage.Client(project=os.environ["GOOGLE_CLOUD_PROJECT"])
        bucket = client.bucket(os.environ["GCS_BUCKET"])
        blob = bucket.blob(key, generation=args.generation) if key else None
        if args.command == "list":
            for page in client.list_blobs(bucket, prefix=prefix, page_size=100).pages:
                for item in page:
                    print(json.dumps({"key": item.name, "generation": item.generation,
                                      "metageneration": item.metageneration, "size": item.size}))
        elif args.command == "put":
            if Path(args.file).stat().st_size > 1024 * 1024:
                raise ValueError("sample limits uploads to 1 MiB")
            blob.upload_from_filename(args.file, if_generation_match=0, checksum="crc32c", timeout=30)
            print(json.dumps({"key": key, "generation": blob.generation,
                              "metageneration": blob.metageneration, "crc32c": blob.crc32c}))
        elif args.command == "get":
            if not args.file:
                raise ValueError("get requires --file (must not exist)")
            with open(args.file, "xb") as output:
                blob.download_to_file(output, checksum="crc32c", timeout=30,
                                      if_generation_match=args.generation)
        elif args.command == "delete":
            blob.delete(if_generation_match=args.generation, timeout=30)
            print(json.dumps({"deleted_generation": args.generation}))
        else:
            # ADC may authenticate reads without implementing the signing interface.
            print(blob.generate_signed_url(version="v4", expiration=300, method="GET",
                                           generation=args.generation))
        return 0
    except GoogleAPICallError as error:
        print(type(error).__name__ + ": GCS request failed", file=sys.stderr)
        return 1
    except Exception as error:
        print(type(error).__name__ + ": check inputs, ADC or signing capability", file=sys.stderr)
        return 1
    finally:
        if client is not None:
            client.close()


if __name__ == "__main__":
    sys.exit(main())

GCS 完整源码使用 list_blobs 的 pages 迭代器，逐页输出 generation、metageneration 和大小；上传附带 if_generation_match=0，删除同时指定 Blob 的 generation 和 if_generation_match。下载先以 xb 打开目标文件，结束或异常都会关闭本地文件；最终 client.close 释放传输资源。成功退出 0，argparse 用法失败退出 2，认证、配置、远端冲突与文件错误退出 1。错误不打印 ADC 内容或签名 URL。

get 允许不传 generation 以展示当前内容，但需要可重复读取时应使用 put 返回的值。若下载失败，本地可能存在部分文件；读者应删除后重试或改成临时文件验证后发布。checksum='crc32c' 只是明确希望 SDK 执行相应校验，特殊下载形式仍须以官方 Blob 文档为准，不能仅因函数返回就宣称业务内容正确。

读者自跑步骤与结果 ​

以下为读者自行运行步骤，本轮均未执行。人工确认实验资源与预算后在仓库根目录建立独立环境，再在当前 shell 设置配置：

sh
python -m venv examples/gcs/.venv
# 激活 .venv；Windows 使用对应 Scripts/Activate.ps1
python -m pip install -r examples/gcs/requirements.txt
# 手工设置项目、桶、前缀以及已获准的 ADC；本项目不自动登录。
python examples/gcs/object_cli.py list
python examples/gcs/object_cli.py put --file ./sample.txt
python examples/gcs/object_cli.py get --key theory-lab/your-session/RETURNED-UUID.bin --generation RETURNED-GENERATION --file ./download.bin
python examples/gcs/object_cli.py sign --key theory-lab/your-session/RETURNED-UUID.bin --generation RETURNED-GENERATION

GCS 上传预期返回 key、generation、metageneration、crc32c；当前为空的前缀 list 无输出且成功。get 保存同一 generation 的字节，读者另外比较原文件与下载文件 SHA-256。sign 在具备可用签名凭据时返回五分钟 V4 URL；只有用户 ADC 时可能失败，这正是认证与签名的差异，不是应该通过提升全项目权限解决的故障。官方 Blob API是这些参数的依据。

清理与练习 ​

GCS 精确删除由读者人工核对 generation 后执行 python examples/gcs/object_cli.py delete --key KEY --generation GENERATION。这次删除的最终状态受版本保留、soft delete 和保留策略影响，不承诺立即释放存储或可以永久恢复。示例不会列举软删除版本、清空桶或修改生命周期；请求、下载流量、旧 generation 和软删除内容可能继续计费。

练习：为何上传必须带 generation=0？答案：UUID 碰撞很低但不是服务端排他条件；generation=0 才表达“只创建新 live 名称”。为何两个重复 put 不构成幂等？答案：程序每次生成新 UUID，条件只约束各自的名称。真正幂等需要业务请求键与对象引用持久化；不应把对象存储条件等同于跨数据库事务。

继续阅读