使用 Python 下載 Blob

發行項
03/17/2024

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

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

必要條件

本文假設您已有專案準備好要使用適用於 Python 的 Azure Blob 儲存體用戶端程式庫。若要了解如何設定您的專案，包括安裝套件、新增 import 陳述式，以及建立授權的用戶端物件，請參閱開始使用 Azure Blob 儲存體和 Python。
若要在程式代碼中使用非同步 API，請參閱非同步程式設計一節中的需求。
授權機制必須具有執行下載作業的權限。若要深入了解，請參閱下列 REST API 作業的授權指導：
- 取得 Blob

下載 Blob

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

BlobClient.download_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：

新增下列匯入陳述式：

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.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())

新增程式碼以下載 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 作業：

取得 Blob (REST API)

程式碼範例

檢視本文中的同步或非同步程式碼範例 (GitHub)

Share via