Azure SDK for Python の一般的な応答の種類について

Azure SDK for Python は、基になる Azure サービス通信プロトコルの呼び出しを抽象化します。そのプロトコルが HTTP であるか AMQP であるか ( ServiceBus や EventHubsなどのメッセージング SDK に使用されます)。たとえば、HTTP を使用するライブラリのいずれかを使用する場合、Azure SDK for Python は HTTP 要求を行い、内部で HTTP 応答を受信します。 SDK は、生の HTTP 応答や JSON ペイロードではなく、直感的な Python オブジェクトを操作できるように、この複雑さを抽象化します。

SDK 操作から受け取るオブジェクトの種類を理解することは、効果的な Azure アプリケーションを作成するために不可欠です。この記事では、発生する一般的な応答の種類と、それらが基になる HTTP 通信にどのように関連しているかを説明します。

注

この記事では、AMQP シナリオではなく、HTTP シナリオのみを調べます。

逆シリアル化された Python オブジェクト

Azure SDK for Python は、サービス操作から厳密に型指定された Python オブジェクトを返すことで、開発者の生産性に優先順位を付けます。 JSON を解析したり、HTTP 状態コードを直接処理したりする代わりに、Azure リソースを Python オブジェクトとして表すリソースモデルを使用します。

たとえば、Azure Storage から BLOB を取得すると、未加工の JSON ディクショナリではなく、BlobProperties、name、sizeなどの属性を持つlast_modified オブジェクトを受け取ります。

from azure.storage.blob import BlobServiceClient

# Connect to storage account
blob_service_client = BlobServiceClient.from_connection_string(connection_string)
container_client = blob_service_client.get_container_client("mycontainer")

# Get blob properties - returns a BlobProperties object
blob_client = container_client.get_blob_client("myblob.txt")
properties = blob_client.get_blob_properties()

# Access properties as Python attributes
print(f"Blob name: {properties.name}")
print(f"Blob size: {properties.size} bytes")
print(f"Last modified: {properties.last_modified}")

データの取得元

データフローを理解すると、SDK がバックグラウンドで行うことを理解するのに役立ちます。

コードで SDK メソッドを呼び出します。get_blob_properties()などのメソッドを呼び出します。
SDK は HTTP 要求を構築します。 SDK は、ヘッダー、認証、およびクエリパラメーターを使用して適切な HTTP 要求をビルドします。
Azure サービスは次のように応答します。 サービスは HTTP 応答を返します。通常は、応答本文に JSON ペイロードが含まれます。
SDK は応答を処理します。 The SDK:
- HTTP 状態コードを確認します。
- 応答本文 (通常は JSON) を解析します。
- 予想されるスキーマに対してデータを検証します。
- データを Python モデルオブジェクトにマップします。
コードは Python オブジェクトを受け取ります。 未加工の HTTP データではなく、逆シリアル化されたオブジェクトを操作します。

この抽象化により、HTTP プロトコルの詳細ではなく、アプリケーションロジックに集中できます。

一般的な応答の種類

Azure SDK for Python では、すべてのサービスで複数の標準応答の種類が使用されます。これらの種類を理解することは、任意の Azure サービスを効果的に操作するのに役立ちます。

リソースモデル

ほとんどの SDK 操作では、リソースモデルが返されます。これらの Python オブジェクトは、Azure リソースを表します。モデルはサービス固有ですが、一貫したパターンに従います。

# Azure Key Vault example
from azure.keyvault.secrets import SecretClient

secret_client = SecretClient(vault_url=vault_url, credential=credential)
secret = secret_client.get_secret("mysecret")  # Returns KeyVaultSecret

print(f"Secret name: {secret.name}")
print(f"Secret value: {secret.value}")
print(f"Secret version: {secret.properties.version}")

# Azure Cosmos DB example
from azure.cosmos import CosmosClient

cosmos_client = CosmosClient(url=cosmos_url, credential=credential)
database = cosmos_client.get_database_client("mydatabase")
container = database.get_container_client("mycontainer")
item = container.read_item(item="item-id", partition_key="partition-value")  # Returns dict

print(f"Item ID: {item['id']}")

コレクション結果の ItemPaged

SDK はリソースを一覧表示すると、改ページ位置を透過的に処理する ItemPaged オブジェクトを返します。

from azure.storage.blob import BlobServiceClient
from azure.core.paging import ItemPaged

blob_service_client = BlobServiceClient.from_connection_string(connection_string)
container_client = blob_service_client.get_container_client("mycontainer")

# list_blobs returns ItemPaged[BlobProperties]
blobs: ItemPaged[BlobProperties] = container_client.list_blobs()

# Iterate naturally - SDK handles pagination
for blob in blobs:
    print(f"Blob: {blob.name}, Size: {blob.size}")

生の HTTP 応答にアクセスする

SDK の高度な抽象化はほとんどのニーズを満たしますが、基になる HTTP 応答へのアクセスが必要な場合があります。一般的なシナリオは、次のとおりです。

失敗した要求のデバッグ。
カスタム応答ヘッダーへのアクセス。
カスタム再試行ロジックの実装。
標準以外の応答形式を扱うこと。

ほとんどの SDK メソッドでは、 raw_response_hook パラメーターを受け取ります。

from azure.keyvault.secrets import SecretClient

secret_client = SecretClient(vault_url=vault_url, credential=credential)

def inspect_response(response):
    # Access the raw HTTP response
    print(f"Request URL: {response.http_request.url}")
    print(f"Status code: {response.http_response.status_code}")
    print(f"Response headers: {dict(response.http_response.headers)}")
    
    # Access custom headers
    request_id = response.http_response.headers.get('x-ms-request-id')
    print(f"Request ID: {request_id}")
    
    # Must return the response
    return response

# Hook is called before deserialization
secret = secret_client.get_secret("mysecret", raw_response_hook=inspect_response)

ページングと反復子

多くの場合、Azure サービスは大量のリソースを返します。 SDK では、 ItemPaged を使用して、すべてを一度にメモリに読み込まずに、これらのコレクションを効率的に処理します。

自動改ページ

反復処理を行うと、SDK によって新しいページが自動的にフェッチされます。

# List all blobs - could be thousands
blobs = container_client.list_blobs()

# SDK fetches pages as needed during iteration
for blob in blobs:
    process_blob(blob)  # Pages loaded on-demand

明示的にページを扱う

必要に応じて、ページを直接操作することもできます。

blobs = container_client.list_blobs()

# Process by page
for page in blobs.by_page():
    print(f"Processing page with {len(list(page))} items")
    for blob in page:
        process_blob(blob)

ページサイズを制御する

多くのリスト操作では、 results_per_page パラメーターを受け入れます。

# Fetch 100 items per page instead of the default
blobs = container_client.list_blobs(results_per_page=100)

一部の Azure サービスの一部の方法には、ページサイズを制御するための他のメカニズムがあります。たとえば、Azure Key Vault と Azure Search では、 top kwarg を使用して、呼び出しごとの結果を制限します。 Azure Search search() メソッドを使用する例については、ソースコードを参照してください。

特殊なケース: 実行時間の長い操作とポーリング

一部の Azure 操作はすぐに完了できません。たとえば、次のようになります。

仮想マシンの作成または削除。
Azure Resource Manager テンプレートのデプロイ。
機械学習モデルのトレーニング。
大容量 BLOB のコピー。

これらの操作は、操作の進行状況を追跡するポーリングオブジェクトを返します。

投票者を操作する

from azure.mgmt.storage import StorageManagementClient

storage_client = StorageManagementClient(credential, subscription_id)

# Start storage account creation
poller = storage_client.storage_accounts.begin_create(
    resource_group_name="myresourcegroup",
    account_name="mystorageaccount",
    parameters=storage_parameters
)

# Option 1: Wait for completion (blocking)
storage_account = poller.result()

# Option 2: Check status periodically
while not poller.done():
    print(f"Status: {poller.status()}")
    time.sleep(5)

storage_account = poller.result()

非同期ポーリング

async/await パターンを使用する場合は、 AsyncLROPollerを操作します。

from azure.storage.blob.aio import BlobServiceClient

async with BlobServiceClient.from_connection_string(connection_string) as client:
    container_client = client.get_container_client("mycontainer")
    
    # Start async copy operation
    blob_client = container_client.get_blob_client("large-blob.vhd")
    poller = await blob_client.begin_copy_from_url(source_url)
    
    # Wait for async completion
    copy_properties = await poller.result()

実行時間の長い操作のオブジェクトのポーリングの例: 仮想マシン

仮想マシンのデプロイは、完了に時間がかかり、ポーリングオブジェクト (同期コードの場合はLROPoller 、非同期コードの場合は AsyncLROPoller ) を返すことによって処理する操作の例です。

from azure.mgmt.compute import ComputeManagementClient
from azure.core.polling import LROPoller

compute_client = ComputeManagementClient(credential, subscription_id)

# Start VM creation - returns immediately with a poller
poller: LROPoller = compute_client.virtual_machines.begin_create_or_update(
    resource_group_name="myresourcegroup",
    vm_name="myvm",
    parameters=vm_parameters
)

# Wait for completion and get the result
vm = poller.result()  # Blocks until operation completes
print(f"VM {vm.name} provisioned successfully")

ページ分割された結果へのアクセス応答

ページングされた結果の場合は、by_page()で raw_response_hook メソッドを使用します。

def page_response_hook(response):
    continuation_token = response.http_response.headers.get('x-ms-continuation')
    print(f"Continuation token: {continuation_token}")
    return response

blobs = container_client.list_blobs()
for page in blobs.by_page(raw_response_hook=page_response_hook):
    for blob in page:
        print(blob.name)

ベストプラクティス

高度な抽象化を優先します。
可能な限り生の応答ではなく、SDK のリソースモデルを操作します。

アンダースコア (_) が付いたメソッドにはアクセスしないでください。 慣例により、これらのメソッドは Python ではプライベートです。パブリック API と比較した破壊的変更などの問題に関する保証はありません。

# Preferred: Work with typed objects
secret = secret_client.get_secret("mysecret")
if secret.properties.enabled:
    use_secret(secret.value)

# Avoid: Manual JSON parsing (unless necessary) ...
# AND avoid accessing any objects or methods that start with `_`
response = secret_client._client.get(...)  # Don't access internal clients
data = json.loads(response.text)
if data['attributes']['enabled']:
    use_secret(data['value'])

改ページ位置を適切に処理します。 リストに変換するのではなく、常にページングされた結果を反復処理します。

# Good: Memory-efficient iteration
for blob in container_client.list_blobs():
    process_blob(blob)

# Avoid: Loading everything into memory
all_blobs = list(container_client.list_blobs())  # Could consume excessive memory

実行時間の長い操作には poller.result() を使用します。 操作が正常に完了することを確認するには、常に result() メソッドを使用します。

# Correct: Wait for operation completion
poller = compute_client.virtual_machines.begin_delete(
    resource_group_name="myresourcegroup",
    vm_name="myvm"
)
poller.result()  # Ensures deletion completes
print("VM deleted successfully")

# Wrong: Assuming immediate completion
poller = compute_client.virtual_machines.begin_delete(...)
print("VM deleted successfully")  # Deletion might still be in progress!

必要な場合にのみ生の応答にアクセスします。 生の応答アクセスを慎重に使用し、特定の要件に対してのみ使用します。

# Good use case: Debugging or logging
def log_request_id(response):
    request_id = response.http_response.headers.get('x-ms-request-id')
    logger.info(f"Operation request ID: {request_id}")
    return response

blob_client.upload_blob(data, raw_response_hook=log_request_id)

# Good use case: Custom error handling
def check_custom_header(response):
    if response.http_response.headers.get('x-custom-error'):
        raise CustomApplicationError("Custom error condition detected")
    return response

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-12-09