使用 Python 上传块 blob

项目
03/22/2024

本文介绍如何使用适用于 Python 的 Azure 存储客户端库来上传 blob。可以通过文件路径、流、二进制对象或文本字符串将数据上传到块 Blob。还可以上传带有索引标记的 Blob。

要了解如何使用异步 API 上传 blob，请参阅异步上传 blob。

先决条件

本文假设已经设置了项目来使用适用于 Python 的 Azure Blob 存储客户端库。若要了解如何设置项目（包括安装包、添加 import 语句和创建授权客户端对象），请参阅开始使用 Azure Blob 存储和 Python。
要通过代码使用异步 API，请参阅异步编程部分中的要求。
授权机制必须具有执行上传操作的权限。若要了解详细信息，请参阅以下 REST API 操作的授权指南：
- 放置 Blob
- 放置块

将数据上传到块 Blob

若要通过流式传输或二进制对象上传 Blob，请使用以下方法：

upload_blob

此方法使用自动分块从数据源创建新的 blob，这意味着数据源可以拆分为更小的区块并上传。若要执行上传，客户端库可以使用 Put Blob 或一系列 Put Block 调用，后跟 Put Block List。此行为取决于对象的总体大小和数据传输选项的设置方式。

通过本地文件路径上传块 Blob

以下示例使用 BlobClient 对象将文件上传到块 blob：

def upload_blob_file(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = container_client.upload_blob(name="sample-blob.txt", data=data, overwrite=True)

通过流式传输上传块 Blob

以下示例创建随机的数据字节，并使用 BlobClient 对象将 BytesIO 对象上传到块 blob：

def upload_blob_stream(self, blob_service_client: BlobServiceClient, container_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    input_stream = io.BytesIO(os.urandom(15))
    blob_client.upload_blob(input_stream, blob_type="BlockBlob")

将二进制数据上传到块 Blob

以下示例使用 BlobClient 对象将二进制数据上传到块 blob：

def upload_blob_data(self, blob_service_client: BlobServiceClient, container_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    data = b"Sample data for blob"

    # Upload the blob data - default blob type is BlockBlob
    blob_client.upload_blob(data, blob_type="BlockBlob")

上传带有索引标记的块 blob

以下示例上传设置了索引标记的块 blob：

def upload_blob_tags(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    sample_tags = {"Content": "image", "Date": "2022-01-01"}
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = container_client.upload_blob(name="sample-blob.txt", data=data, tags=sample_tags)

使用配置选项上传块 blob

上传 blob 时，可以定义客户端库配置选项。可以优化这些选项以提高性能、增强可靠性和优化成本。以下代码示例演示在实例化 BlobClient 时如何在方法级别和客户端级别为上传定义配置选项。还可以为 ContainerClient 实例或 BlobServiceClient 实例配置这些选项。

指定用于上传的数据传输选项

可以在实例化客户端时设置配置选项，来优化数据传输操作的性能。在 Python 中构造客户端对象时，可传递以下关键字参数：

max_block_size - 上传块 blob 的最大区块大小（以区块为单位）。默认值为 4 MiB。
max_single_put_size - 如果 blob 大小小于或等于 max_single_put_size，则使用单个 Put Blob 请求上传 blob。如果 blob 大小大于 max_single_put_size 或未知，则使用 Put Block 上传 blob（以区块为单位）并使用 Put Block List 提交 blob。默认值为 64 MiB。

若要详细了解 Blob 存储的传输大小限制，请参阅 Blob 存储的缩放目标。

对于上传操作，还可以在调用 upload_blob 时传递 max_concurrency 参数。此参数定义当 blob 大小超过 64 MiB 时要使用的最大并行连接数。

以下代码示例演示如何在创建 BlobClient 对象时指定数据传输选项，以及如何使用该客户端对象上传数据。此示例中提供的值不作为建议。若要正确优化这些值，需要考虑应用的特定需求。

def upload_blob_transfer_options(self, account_url: str, container_name: str, blob_name: str):
    # Create a BlobClient object with data transfer options for upload
    blob_client = BlobClient(
        account_url=account_url, 
        container_name=container_name, 
        blob_name=blob_name,
        credential=DefaultAzureCredential(),
        max_block_size=1024*1024*4, # 4 MiB
        max_single_put_size=1024*1024*8 # 8 MiB
    )
    
    with open(file=os.path.join(r'file_path', blob_name), mode="rb") as data:
        blob_client = blob_client.upload_blob(data=data, overwrite=True, max_concurrency=2)

若要详细了解如何优化数据传输选项，请参阅使用 Python 对上传和下载进行性能优化。

在上传期间设置 blob 的访问层

通过将 standard_blob_tier 关键字参数传递给 upload_blob，可以在上传时设置 blob 的访问层。 Azure 存储提供不同的访问层，以便你可以基于使用方式，以最具成本效益的方式存储 blob 数据。

以下代码示例演示如何在上传 blob 时设置访问层：

def upload_blob_access_tier(self, blob_service_client: BlobServiceClient, container_name: str, blob_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
    
    #Upload blob to the cool tier
    with open(file=os.path.join(r'file_path', blob_name), mode="rb") as data:
        blob_client = blob_client.upload_blob(data=data, overwrite=True, standard_blob_tier=StandardBlobTier.COOL)

仅允许为块 blob 设置访问层。可以将块 blob 的访问层设置为 Hot、Cool、Cold 或 Archive。若要将访问层设置为 Cold，必须至少使用客户端库版本 12.15.0。

若要详细了解访问层，请参阅访问层概述。

通过暂存块和提交上传块 Blob

可以通过手动暂存单个数据块来更好地控制将上传内容划分为块的方式。暂存构成 Blob 的所有块后，可以将其提交到 Blob 存储。

使用以下方法创建要作为 Blob 的一部分提交的新块：

stage_block

使用以下方法通过指定构成 blob 的块 ID 列表来写入 Blob：

commit_block_list

以下示例从文件读取数据，并暂存要作为 Blob 的一部分提交的块：

def upload_blocks(self, blob_container_client: ContainerClient, local_file_path: str, block_size: int):
    file_name = os.path.basename(local_file_path)
    blob_client = blob_container_client.get_blob_client(file_name)

    with open(file=local_file_path, mode="rb") as file_stream:
        block_id_list = []

        while True:
            buffer = file_stream.read(block_size)
            if not buffer:
                break

            block_id = uuid.uuid4().hex
            block_id_list.append(BlobBlock(block_id=block_id))

            blob_client.stage_block(block_id=block_id, data=buffer, length=len(buffer))

        blob_client.commit_block_list(block_id_list)

异步上传 Blob

适用于 Python 的 Azure Blob 存储客户端库支持异步上传 Blob。要详细了解项目设置要求，请参阅异步编程。

按照以下步骤使用异步 API 上传 Blob：

添加以下 import 语句：

import asyncio

from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient, BlobClient, ContainerClient

添加代码以使用asyncio.run运行程序。此函数运行传递的协同例程（我们示例中的main()），并管理asyncio事件循环。使用 async/await 语法声明协同例程。在此示例中，main()协同例程先使用async with创建顶级BlobServiceClient，然后调用上传 blob 的方法。请注意，只有顶级客户端需要使用async with，因为基于它创建的其他客户端会共享同一连接池。

async def main():
    sample = BlobSamples()

    # TODO: Replace <storage-account-name> with your actual storage account name
    account_url = "https://<storage-account-name>.blob.core.chinacloudapi.cn"
    credential = DefaultAzureCredential()

    async with BlobServiceClient(account_url, credential=credential) as blob_service_client:
        await sample.upload_blob_file(blob_service_client, "sample-container")

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

添加代码以上传 Blob。以下示例使用ContainerClient对象从本地文件路径上传 blob。该代码与同步示例相同，除了该方法使用async关键字声明，await关键字在调用upload_blob方法时使用。

async def upload_blob_file(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = await container_client.upload_blob(name="sample-blob.txt", data=data, overwrite=True)

有了此基本设置，即可使用 async/await 语法将本文中的其他示例实现为协同例程。

资源

若要详细了解如何使用适用于 Python 的 Azure Blob 存储客户端库来上传 blob，请参阅以下资源。

REST API 操作

Azure SDK for Python 包含基于 Azure REST API 而生成的库，允许你通过熟悉的 Python 范例与 REST API 操作进行交互。用于上传 blob 的客户端库方法使用以下 REST API 操作：