共用方式為

使用 Python 下載 Blob

  • 發行項

本文說明如何使用適用於 Python 的 Azure 儲存體用戶端程式庫來下載 Blob。 您可以將 Blob 資料下載到各種目的地，包括本機檔案路徑、資料流或文字字串。 您也可以開啟 Blob 資料流並從中讀取。

若要了解如何使用非同步 API 下載 Blob，請參閱以非同步方式下載 Blob。

必要條件

設定您的環境

如果沒有現有的專案，本章節會說明如何設定專案以使用適用於 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.windows.net"
credential = DefaultAzureCredential()

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

您也可以直接或從 BlobServiceClient 物件建立特定容器或 Blob 的用戶端物件。 若要深入了解如何建立及管理用戶端物件，請參閱建立和管理與資料資源互動的用戶端端物件 (部分機器翻譯)。

下載 Blob

您可以使用下列方法來下載 Blob：

download_blob 方法會傳回 StorageStreamDownloader 物件。 下載期間，客戶端程式庫會將下載要求分割成區塊，其中每個區塊都會以個別的取得 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 內容下載為文字。 在此範例中，encoding 參數是 readall() 傳回字元串的必要參數，否則會傳回位元組：

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：

  1. 新增下列匯入陳述式：

    import asyncio

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

  2. 新增程式碼以使用 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.windows.net"
    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())

  3. 新增程式碼以下載 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，請參閱下列資源。

程式碼範例

REST API 操作

Azure SDK for Python 包含建置在 Azure REST API 之上的程式庫，可讓您透過熟悉的 Python 範例與 REST API 作業進行互動。 用來下載 Blob 的用戶端程式庫方法會使用下列 REST API 作業：

用戶端程式庫資源

相關內容

  • 本文是適用於 Python 的 Blob 儲存體開發人員指南的一部分。 若要深入了解，請參閱 建置 Python 應用程式 中的開發人員指南文章完整清單。