Python 异步 Azure Blob 上传

Question

我有一个代码可以读取文件目录并将文件上传到 Azure Blob 存储。它运行良好并且文件上传成功。但是，我需要帮助来修改此代码以运行异步操作以实现并发上传。

import os
import asyncio
import yaml
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__


def load_config():
    path_root = os.path.dirname(os.path.abspath(__file__))
    with open(path_root + "/config.yaml", "r") as configfile:
        return yaml.load(configfile, Loader=yaml.FullLoader)


def read_files(dir):
    with os.scandir(dir) as files:
        for filename in files:
            if filename.is_file() and not filename.name.startswith('.'):
                yield filename


def upload(files, connection_string, container_name):
    container_client = ContainerClient.from_connection_string(connection_string, container_name)
    print("Uploading images to remote blob storage")

    for file in files:
        blob_client = container_client.get_blob_client(file.name)
        with open(file.path, "rb") as data:
            blob_client.upload_blob(data)
            print(f"{file.name} upload to remote blob storage")


config = load_config()
images = read_files(config['source_folder'] + '/images')
upload(images, config['azure_storage_connectionstring'], config['images_container_name'])

Answer 1

您可以使用该包azure.storage.blob.aio异步上传 blob。

例如

async def load_config():
    path_root = os.path.dirname(os.path.abspath(__file__))
    with open(path_root + "/config.yaml", "r") as configfile:
        return yaml.load(configfile, Loader=yaml.FullLoader)


async def read_files(dir):
    with os.scandir(dir) as files:
        for filename in files:
            if filename.is_file() and not filename.name.startswith('.'):
                yield filename


async def upload_blob():
    tasks = []
    config = await load_config()
    container_client = ContainerClient.from_connection_string(
        config['azure_storage_connectionstring'],config['images_container_name'])
    async with container_client:
        async for file in read_files(config['source_folder'] + '/images'):
            with open(file.path, "rb") as data:
                tasks.append(asyncio.create_task(
                    container_client.upload_blob(name=file.name, data=data)))
                print(f"{file.name} upload to remote blob storage")

                await asyncio.gather(*tasks)
    print("Finished")
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(upload_blob())