python - 从源容器复制 blob 并异步上传到目标容器

Question

我正在尝试使用 azure.storage.blob.aio 将一个容器中的 blob 上传到另一个容器，但是如果函数按预期工作，我会收到警告消息:

Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f6a2ebf6e50>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7f6a2e3553a0>, 32966.677798886)]']
connector: <aiohttp.connector.TCPConnector object at 0x7f6a2ebf6eb0>
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f6a2ec03400>
Unclosed connector
...

我的代码有什么问题，我该如何解决以上消息？

async def upload_blob(blob_args):
    # srource blob
    src_container_client = ContainerClient.from_connection_string(blob_args["src_connect_str"], blob_args["src_container"])

    # destination blob
    target_container_client = ContainerClient.from_connection_string(blob_args["target_connect_str"], blob_args["target_container"])
  
    tasks = []
    print(f"\nReading blobs from the container:")
    async for source_blob in src_container_client.list_blobs():
        # push data into specified stream
        source_blob_client = src_container_client.get_blob_client(source_blob.name)
        stream_downloader = await source_blob_client.download_blob()
        stream = await stream_downloader.readall()
            
        # create the BlobClient from the ContainerClient to interact with a specific blob
        target_blob_client = target_container_client.get_blob_client(source_blob.name)

        print(f"\t- Transfering {source_blob.name}.")
        tasks.append(asyncio.create_task(target_blob_client.upload_blob(stream)))

    await asyncio.gather(*tasks)
    print("Finished")

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(upload_blob(blob_args))

Answer 1

关于警告，您没有使用 ContainerClient 运行异步上下文管理器。

例如

import io
from azure.storage.blob.aio import ContainerClient
import asyncio


async def copy_blobs():
    # srource blob
    source_container_client = ContainerClient.from_connection_string(
        conn_str, 'upload')

    # destination blob
    target_container_client = ContainerClient.from_connection_string(
        conn_str, 'copy')
    async with source_container_client, target_container_client:
        tasks = []
        print(f"\nReading blobs from the container:")
        async for source_blob in source_container_client.list_blobs():
            source_blob_client = src_container_client.get_blob_client(source_blob.name)
            stream_downloader = await source_blob_client.download_blob()
            stream = await stream_downloader.readall()
            
            tasks.append(asyncio.create_task(target_blob_client.upload_blob(stream)))
            
            print(f"\t- Transfering {source_blob.name}.")

        await asyncio.gather(*tasks)
        print("Finished")

if __name__ == "__main__":
    asyncio.set_event_loop(asyncio.new_event_loop())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(copy_blobs())