Python异步操作Google Cloud Storage：文件列举、下载、删除实现

你现有代码已经通过本地creds.json服务凭证完成了鉴权配置，直接基于同个客户端实例补充列表查询、文件下载、文件删除逻辑即可，不需要额外修改鉴权部分。

权限前置检查 #

• 确保creds.json对应的服务账号已被授予目标存储桶的以下权限，否则对应接口会返回权限错误：
  • storage.objects.list：查询文件列表
  • storage.objects.get：下载文件
  • storage.objects.delete：删除文件

核心功能实现代码 #

三个新增功能均为异步实现，和你现有代码的异步模型完全兼容：

import asyncio
import aiohttp
from aiofile import AIOFile
from gcloud.aio.storage import Storage 


BUCKET_NAME = 'unscanned-malware-scann-lab'
FILE_NAME  = 'testeArquivoUp.txt'

# 原有上传逻辑保留
async def async_upload_to_bucket(blob_name, file_obj, folder='uploads'):
    """上传文件到存储桶指定目录"""
    async with aiohttp.ClientSession() as session:
        storage = Storage(service_file='./creds.json', session=session) 
        status = await storage.upload(BUCKET_NAME, f'{folder}/{blob_name}', file_obj)
        return status['selfLink']

# 新增：查询文件列表
async def list_blobs_in_bucket(folder=None):
    """
    查询存储桶下的文件列表
    传入folder参数时，仅返回对应目录下的文件（GCS通过前缀匹配实现目录逻辑）
    """
    async with aiohttp.ClientSession() as session:
        storage = Storage(service_file='./creds.json', session=session)
        list_params = {}
        if folder:
            list_params['prefix'] = f'{folder.rstrip("/")}/'
        resp = await storage.list_objects(BUCKET_NAME, params=list_params)
        # 提取核心字段返回
        result = []
        for blob in resp.get('items', []):
            result.append({
                "blob_path": blob["name"],
                "size_byte": int(blob["size"]),
                "update_time": blob["updated"],
                "download_link": blob["mediaLink"]
            })
        return result

# 新增：下载文件
async def download_blob(blob_path, local_save_path):
    """
    下载桶内指定文件到本地
    :param blob_path: 桶内文件完整路径，例：uploads/test.txt
    :param local_save_path: 本地文件保存路径
    """
    async with aiohttp.ClientSession() as session:
        storage = Storage(service_file='./creds.json', session=session)
        content = await storage.download(BUCKET_NAME, blob_path)
        async with AIOFile(local_save_path, "wb") as afp:
            await afp.write(content)
        return local_save_path

# 新增：删除文件
async def delete_blob(blob_path):
    """
    删除桶内指定文件
    :param blob_path: 桶内文件完整路径，例：uploads/test.txt
    """
    async with aiohttp.ClientSession() as session:
        storage = Storage(service_file='./creds.json', session=session)
        return await storage.delete(BUCKET_NAME, blob_path)

调用示例 #

把原有main函数替换为以下内容即可测试全流程：

async def main():
    print('开始执行GCS操作')
    # 原有上传流程
    async with AIOFile(FILE_NAME, mode='r') as afp:
        file_content = await afp.read()
        upload_url = await async_upload_to_bucket(FILE_NAME, file_content)
        print(f'上传完成，文件自链接：{upload_url}')

    # 查询uploads目录下所有文件
    uploads_files = await list_blobs_in_bucket(folder="uploads")
    print("\nuploads目录文件列表：")
    for f in uploads_files:
        print(f"- {f['blob_path']} | 大小：{f['size_byte']}字节 | 更新时间：{f['update_time']}")

    # 下载刚上传的文件到本地
    target_blob = f"uploads/{FILE_NAME}"
    saved_path = await download_blob(target_blob, f"./downloaded_{FILE_NAME}")
    print(f"\n下载完成，本地保存路径：{saved_path}")

    # 测试删除功能时可放开以下注释
    # del_resp = await delete_blob(target_blob)
    # print(f"\n删除操作返回：{del_resp}")

if __name__ == "__main__":
    asyncio.run(main())

补充说明 #

• 不需要指定目录、要查询桶根目录所有文件时，直接调用list_blobs_in_bucket()不传folder参数即可
• 处理超过100MB的大文件时，建议给upload/download方法传入分块大小参数，避免一次性占满内存
• 运行环境需要能正常连通GCS服务，本地调试可按需配置系统代理

内容的提问来源于stack exchange，提问作者Rodrigo Augusto Martins