如何用Python将Google Storage可读签名URL流传输至另一存储桶

方案一：使用Google Cloud Storage客户端库直接流式中转 #

这个方案借助官方客户端库的流式API，直接从源签名URL读取数据流并写入目标桶，全程无需本地存储文件，是最简洁的实现方式。

依赖安装 #

pip install google-cloud-storage requests

代码实现 #

import requests
from google.cloud import storage

def stream_transfer_from_signed_url(source_signed_url: str, dest_bucket_name: str, dest_blob_name: str):
    # 初始化GCS客户端
    storage_client = storage.Client()
    dest_bucket = storage_client.bucket(dest_bucket_name)
    dest_blob = dest_bucket.blob(dest_blob_name)

    # 从源签名URL流式读取数据
    with requests.get(source_signed_url, stream=True) as response:
        response.raise_for_status()
        # 流式写入目标GCS对象
        dest_blob.upload_from_file(response.raw, content_type=response.headers.get('Content-Type'))

# 使用示例
if __name__ == "__main__":
    SOURCE_SIGNED_URL = "你的源可读签名URL"
    DEST_BUCKET = "目标存储桶名称"
    DEST_BLOB_NAME = "目标文件路径/文件名"
    stream_transfer_from_signed_url(SOURCE_SIGNED_URL, DEST_BUCKET, DEST_BLOB_NAME)

优势 #

• 官方库封装所有GCS交互细节，无需手动处理签名逻辑
• 全程流式处理，内存占用极低，适配大文件转存
• 自动处理HTTP错误与基础重试逻辑

方案二：生成目标桶上传签名URL，通过HTTP流式传输 #

如果需要脱离GCS客户端库（比如轻量无服务环境），可以生成目标桶的上传签名URL，直接通过HTTP请求转发源数据流。

代码实现 #

import requests
from google.cloud import storage
from datetime import timedelta

def generate_upload_signed_url(bucket_name: str, blob_name: str) -> str:
    # 生成有效期1小时的目标桶上传签名URL
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    return blob.generate_signed_url(
        version="v4",
        expiration=timedelta(hours=1),
        method="PUT",
        content_type="application/octet-stream"  # 可根据实际文件类型调整
    )

def stream_transfer_via_signed_urls(source_signed_url: str, dest_upload_url: str):
    # 流式读取源URL数据并转发到目标上传URL
    with requests.get(source_signed_url, stream=True) as source_resp:
        source_resp.raise_for_status()
        headers = {"Content-Type": source_resp.headers.get('Content-Type', 'application/octet-stream')}
        with requests.put(dest_upload_url, data=source_resp.raw, headers=headers) as dest_resp:
            dest_resp.raise_for_status()

# 使用示例
if __name__ == "__main__":
    SOURCE_SIGNED_URL = "你的源可读签名URL"
    DEST_BUCKET = "目标存储桶名称"
    DEST_BLOB_NAME = "目标文件路径/文件名"
    
    dest_upload_url = generate_upload_signed_url(DEST_BUCKET, DEST_BLOB_NAME)
    stream_transfer_via_signed_urls(SOURCE_SIGNED_URL, dest_upload_url)

适用场景 #

• 轻量函数等需精简依赖的环境
• 跨账号/跨项目转存，且目标桶允许通过签名URL上传

注意事项 #

• 权限要求：方案一需要目标桶的storage.objects.create权限；方案二需要目标桶的storage.objects.create权限以生成上传签名URL
• 大文件优化：可在requests.get中添加chunk_size=1024*1024调整流式块大小（默认128KB）
• 异常处理：可根据业务需求添加重试逻辑，比如用tenacity库实现失败重试

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