如何使用gcloud.aio.storage向新Bucket上传Blob？

我正在使用gcloud.aio.storage搭配fake-gcs-server模拟器，已成功向现有Bucket上传Blob，测试代码及结果如下：

import os
import pytest
# Set ENV before Storage import, otherwise it gets set to prod gcs endpoint
os.environ["STORAGE_EMULATOR_HOST"] = "localhost:4443"
from gcloud.aio.storage import Storage

### WORKS OK #####
@pytest.mark.asyncio
async def test_upload_blob_to_existing_bucket() -> None:

    async with Storage() as async_client:
        existing_bucket_name: str = "fake_server_bucket"
        await async_client.upload(existing_bucket_name, "myBlobName.json", "blobContent")

执行测试结果：

pytest test_gcs_async_upload.py
======= 1 passed in 0.24s ======

现在我需要向新Bucket上传Blob，但以下代码始终返回Bucket不存在的404错误：aiohttp.client_exceptions.ClientResponseError: 404, message='Not Found...'

我查看了gcloud.aio.storage的测试用例，未发现操作差异，且该库没有同步版google.cloud.storage中的create_bucket(bucket)创建方法。

测试代码及错误如下：

import os
import pytest
# Set ENV before Storage import, otherwise it gets set to prod gcs endpoint
os.environ["STORAGE_EMULATOR_HOST"] = "localhost:4443"
from gcloud.aio.storage import Storage

### Returns 404 notfound error ###
@pytest.mark.asyncio
async def test_upload_blob_to_new_bucket() -> None:

    async with Storage() as async_client:
        # async_client.create_bucket(bucket) #Cannot do this in gcloud.aio.storage lib!
        new_bucket_name: str = "newbucket"
        await async_client.upload(new_bucket_name, "myBlobName.json", "blobContent")

错误输出：

pytest test_gcs_async_upload.py

E   aiohttp.client_exceptions.ClientResponseError: 404, message='Not Found: 
{"error":{"code":404,"message":"Not Found","errors":null}}
', url=URL('http://localhost:4443/upload/storage/v1/b/newbucket/o? name=myBlobName.json&uploadType=media')
======= 1 failed in 0.20s ======

环境配置：

python = ">=3.8,<3.11"
gcloud-aio-storage = "^7.0.1"
pytest = "^7.1.2"
pytest-asyncio = "^0.19.0"

fake-gcs-server is LATEST, TAG = 1 digest = 80e798b48cdd.
It runs as a docker instance and is prepopulated with a bucket and some blobs at startup  

解决方案 #

gcloud.aio.storage确实没有直接暴露create_bucket方法，但可以通过两种方式解决：

方法1：调用底层HTTP API创建Bucket

利用Storage类的request方法直接调用GCS的Bucket创建接口，代码示例：

import os
import pytest
os.environ["STORAGE_EMULATOR_HOST"] = "localhost:4443"
from gcloud.aio.storage import Storage

@pytest.mark.asyncio
async def test_upload_blob_to_new_bucket() -> None:
    async with Storage() as async_client:
        new_bucket_name = "newbucket"
        # 先创建Bucket
        await async_client.request(
            method='POST',
            path='/b',
            json={'name': new_bucket_name}
        )
        # 再上传Blob
        await async_client.upload(new_bucket_name, "myBlobName.json", "blobContent")

方法2：启动模拟器时预创建Bucket

如果测试场景固定，可在启动fake-gcs-server的Docker命令中添加-bucket参数提前创建目标Bucket：

docker run -d -p 4443:4443 fsouza/fake-gcs-server:latest -bucket fake_server_bucket -bucket newbucket

补充说明

• GCS及fake-gcs-server均遵循"先创建Bucket再上传Blob"的规则，直接上传到不存在的Bucket必然返回404
• gcloud.aio.storage的request方法可调用任意GCS REST API，用于弥补库中未封装的功能

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