使用 Python 下载 Blob

项目
10/15/2024

本文介绍了如何使用适用于 Python 的 Azure 存储客户端库来下载 blob。可以将 Blob 数据下载到各种目标，包括本地文件路径、流或文本字符串。还可以打开 Blob 流并从中读取。

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

先决条件

Azure 订阅 - 创建试用订阅。
Azure 存储帐户 - 创建存储帐户
Python 3.8+

设置你的环境

如果没有现有项目，请查看本部分，其中介绍如何设置项目来使用适用于 Python 的 Azure Blob 存储客户端库。有关更多详细信息，请参阅 Azure Blob 存储和 Python 入门。

要使用本文中的代码示例，请按照以下步骤设置项目。

安装包

使用 pip install 安装以下包：

pip install azure-storage-blob azure-identity

添加 import 语句

添加以下 import 语句：

import io
import os
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient, BlobClient

授权

授权机制必须具有执行下载操作所需的权限。若要使用 Microsoft Entra ID 进行授权（建议），需有 Azure RBAC 内置角色“存储 Blob 数据读取者”或更高级别的角色。若要了解详细信息，请参阅获取 Blob (REST API) 的授权指南。

创建客户端对象

若要将应用连接到 Blob 存储，请创建 BlobServiceClient 的实例。以下示例演示如何使用 DefaultAzureCredential 创建客户端对象进行授权：

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

# Create the BlobServiceClient object
blob_service_client = BlobServiceClient(account_url, credential=credential)

还可以为特定容器或 Blob 创建客户端对象，不管是直接创建还是通过 BlobServiceClient 对象创建。要详细了解如何创建和管理客户端对象，请参阅创建和管理与数据资源交互的客户端对象。

下载 Blob

可以使用以下方法下载 Blob：

BlobClient.download_blob

download_blob 方法返回一个 StorageStreamDownloader 对象。在下载过程中，客户端库将下载请求拆分为多个区块，其中每个区块都通过单独的 Get Blob 范围请求进行下载。此行为取决于 blob 的总大小以及数据传输选项的设置方式。

下载到文件路径

以下示例将 Blob 下载到一个文件路径：

def download_blob_to_file(self, blob_service_client: BlobServiceClient, container_name):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    with open(file=os.path.join(r'filepath', 'filename'), mode="wb") as sample_blob:
        download_stream = blob_client.download_blob()
        sample_blob.write(download_stream.readall())

下载到流

以下示例将 Blob 下载到一个流。在此示例中，StorageStreamDownloader.read_into 将 blob 内容下载到一个流并返回读取的字节数：

def download_blob_to_stream(self, blob_service_client: BlobServiceClient, container_name):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")

    # readinto() downloads the blob contents to a stream and returns the number of bytes read
    stream = io.BytesIO()
    num_bytes = blob_client.download_blob().readinto(stream)
    print(f"Number of bytes: {num_bytes}")

在区块中下载 Blob

以下示例下载 Blob 并循环访问下载流中的区块。在此示例中，StorageStreamDownloader.chunks 返回一个迭代器，这允许你在区块中读取 blob 内容：

def download_blob_chunks(self, blob_service_client: BlobServiceClient, container_name):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")

    # This returns a StorageStreamDownloader
    stream = blob_client.download_blob()
    chunk_list = []

    # Read data in chunks to avoid loading all into memory at once
    for chunk in stream.chunks():
        # Process your data (anything can be done here - 'chunk' is a byte array)
        chunk_list.append(chunk)

下载到字符串

以下示例将 Blob 内容下载为文本。在此示例中，要想让 readall() 返回字符串，必须使用 encoding 参数，否则它将返回字节：

def download_blob_to_string(self, blob_service_client: BlobServiceClient, container_name):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")

    # encoding param is necessary for readall() to return str, otherwise it returns bytes
    downloader = blob_client.download_blob(max_concurrency=1, encoding='UTF-8')
    blob_text = downloader.readall()
    print(f"Blob contents: {blob_text}")

使用配置选项下载块 blob

下载 blob 时，可以定义客户端库配置选项。可以对这些选项进行调整，以提高性能和可靠性。以下代码示例演示在实例化 BlobClient 时如何在方法级别和客户端级别为下载定义配置选项。还可以为 ContainerClient 实例或 BlobServiceClient 实例配置这些选项。

在下载时指定数据传输选项

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

max_chunk_get_size - 用于下载 blob 的最大区块大小。默认值为 4 MiB。
max_single_get_size - 要在单个调用中下载的 blob 的最大大小。如果 blob 总大小超过 max_single_get_size，将分区块下载其余 blob 数据。默认值为 32 MiB。

对于下载操作，还可以在调用 download_blob 时传递 max_concurrency 参数。此参数定义下载操作的最大并行连接数。

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

def download_blob_transfer_options(self, account_url: str, container_name: str, blob_name: str):
    # Create a BlobClient object with data transfer options for download
    blob_client = BlobClient(
        account_url=account_url, 
        container_name=container_name, 
        blob_name=blob_name,
        credential=DefaultAzureCredential(),
        max_single_get_size=1024*1024*32, # 32 MiB
        max_chunk_get_size=1024*1024*4 # 4 MiB
    )

    with open(file=os.path.join(r'file_path', 'file_name'), mode="wb") as sample_blob:
        download_stream = blob_client.download_blob(max_concurrency=2)
        sample_blob.write(download_stream.readall())

异步下载 blob

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

按照以下步骤使用异步 API 下载 blob：

添加以下 import 语句：

import asyncio

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

添加代码以使用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.download_blob_to_file(blob_service_client, "sample-container")

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

添加代码以下载 blob。以下示例使用 BlobClient 对象将 blob 下载到本地文件路径。该代码与同步示例相同，除了该方法使用async关键字声明，await关键字在调用download_blob方法时使用。

async def download_blob_to_file(self, blob_service_client: BlobServiceClient, container_name):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    with open(file=os.path.join(r'filepath', 'filename'), mode="wb") as sample_blob:
        download_stream = await blob_client.download_blob()
        data = await download_stream.readall()
        sample_blob.write(data)

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

资源

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

代码示例

查看本文中的同步或异步代码示例 (GitHub)

REST API 操作

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

获取 Blob (REST API)

客户端库资源

本文是 Python 版 Blob 存储开发人员指南的一部分。若要了解详细信息，请参阅生成 Python 应用中的开发人员指南文章的完整列表。

通过