时间戳

首页 > 解决方案 > GCP & GPG - 从云存储流式传输文件并上传到云存储

python - GCP & GPG - 从云存储流式传输文件并上传到云存储

问题描述

这与之前提出的这个问题有点相似：gnupg - decrypt into Python bytesio stream

我需要从 GCP Cloud Storage 流式传输文件，使用 GPG 在流中加密它们，同时将加密的数据上传到 GCP Cloud Storage。这可能吗？

阅读 API 文档（https://googleapis.dev/python/storage/latest/blobs.html），我可以使用以下方式流式传输文件：

from google.cloud import storage
client = storage.Client()
bucket = client.bucket("bucket-name")
blob = bucket.get_blob("blob-name.txt")
with blob.open("rt") as f:
    print(f.read())

我可以使用以下方法加密本地文件：

with open('Test.txt', 'rb') as f:
                    status = gpg.encrypt_file(
                        f,sign=public_key_fingerprint,
                        recipients=private_key_recipient,
                        passphrase=private_key_pass,
                        always_trust=True,
                        output = output_file
                    )

我可以在加密期间使用以下命令将 GCP 文件流式传输到本地 .gpg 文件：

with blob.open("rb") as f:
             status = gpg.encrypt_file(
                    f,sign=public_key_fingerprint,
                    recipients=private_key_recipient,
                    passphrase=private_key_pass,
                    always_trust=True,
                    output = output_file
                )

但我需要输出是 Cloud Storage 中的文件，而不是本地系统。

我试过了：

bucket = storage_client.bucket(input_bucket) 
blob = bucket.get_blob(input_file)
uploadbucket = storage_client.bucket(output_bucket)
uploadblob = bucket.blob(output_file)
with blob.open("rt") as f:
    with uploadblob.open("wt") as en:
        status = gpg.encrypt_file(
            f,sign=public_key_fingerprint,
            recipients=private_key_recipient,
            passphrase=private_key_pass,
            always_trust=True,
            output = en
        )

但这给了我错误：

path should be string, bytes, os.PathLike or integer, not TextIOWrapper

对此的任何帮助将不胜感激！

标签： pythongoogle-cloud-platformgnupg

解决方案

问题在于gpg.encrypt_file(..., output=en)通话。output参数必须是带有内容目的地的字符串。

您可以gpg.encrypt()改用，然后将内容写入 blob，如下所示：

with blob.open("rt") as f:
    with uploadblob.open("wt") as en:
        encrypted = gpg.encrypt(
            f,sign=public_key_fingerprint,
            recipients=private_key_recipient,
            passphrase=private_key_pass,
            always_trust=True,
            armor=True)
        en.write(encrypted)

（我已经armor=True明确地记住输出是 ascii 装甲输出，而不是二进制）。

推荐阅读