Python を使用して BLOB をダウンロードする

この記事では、Python 用の Azure Storage クライアント ライブラリを使用して BLOB をダウンロードする方法について説明します。 BLOB データは、ローカル ファイル パス、ストリーム、テキスト文字列など、さまざまな宛先にダウンロードできます。 BLOB ストリームを開き、そこから読み取ることもできます。

非同期 API を使用した BLOB のダウンロードの詳細については、「BLOB を非同期にダウンロードする」に関する記事を参照してください。

前提条件

• Azure サブスクリプション - 無料アカウントを作成する
• Azure Storage アカウント - ストレージ アカウントの作成
• python=3.8.10

環境を設定する

既存のプロジェクトがない場合、Python 用 Azure Blob Storage クライアント ライブラリを操作するためのプロジェクトをセットアップする方法についてこのセクションで説明します。 詳細については、「Azure Blob Storage と 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 組み込みロールの Storage BLOB データ閲覧者以上が必要です。 詳細については、Get Blob (REST API) に関するページの認可のガイダンスを参照してください。

クライアント オブジェクトの作成

アプリを Blob Storage に接続するには、 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)

また、特定のコンテナーまたは 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 - 1 回の呼び出しでダウンロードされる 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 Storage クライアント ライブラリでは、BLOB の非同期ダウンロードがサポートされています。 プロジェクトのセットアップ要件の詳細については、「非同期プログラミング」を参照してください。

非同期 API を使用して BLOB をダウンロードするには、次の手順に従います。

1. 次の import ステートメントを追加します。

   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 キーワードを使用してメソッドが宣言され、download_blob メソッドを呼び出すときに await キーワードが使用される点が異なります。

   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 Storage クライアント ライブラリを使用して BLOB をダウンロードする方法について詳しくは、次のリソースを参照してください。

コード サンプル

• この記事の同期または非同期のサンプル コードを表示する (GitHub)

REST API の操作

Azure SDK for Python には Azure REST API に基づき構築されたライブラリが含まれるため、使い慣れた Python パラダイムを通じて REST API 操作を実施できます。 BLOB をダウンロードするためのクライアント ライブラリ メソッドは、次の REST API 操作を使用します。

• Get Blob (REST API)

クライアント ライブラリのリソース

• クライアント ライブラリのリファレンス ドキュメント
• クライアント ライブラリ ソース コード
• パッケージ (PyPi)

関連するコンテンツ

• この記事は、Python の Blob Storage 開発者ガイドの一部です。 詳細については、「Python アプリの構築」にある開発者ガイドの記事の完全な一覧を参照してください。