了解 Azure SDK for Python 中常見的回應類型

Python 的 Azure SDK 抽象了對底層 Azure 服務通訊協定的呼叫，無論該協定是 HTTP 或 AMQP（用於訊息 SDK 如 ServiceBus 和 EventHubs）。 舉例來說，如果你使用使用 HTTP 的函式庫，Azure SDK for Python 會發出 HTTP 請求並接收 HTTP 回應。 SDK 抽象化了這些複雜度，讓你能使用直覺式的 Python 物件，而非原始的 HTTP 回應或 JSON 負載。

了解你從 SDK 操作中接收到的物件類型，對於撰寫有效的 Azure 應用程式至關重要。 本文將說明你常見的回應類型，以及它們如何與底層的 HTTP 通訊相關。

備註

本文僅探討 HTTP 情境，而非 AMQP 情境。

還原序列化的 Python 物件

適用於 Python 的 Azure SDK 會從服務作業傳回強型別 Python 物件，以排定開發人員生產力的優先順序。 您不需要剖析 JSON 或直接處理 HTTP 狀態碼，而是使用將 Azure 資源表示為 Python 物件的資源模型。

例如，當你從 Azure Storage 取得一個 blob 時，你會收到一個具有 name、size 和 last_modified 屬性的 BlobProperties 物件，而不是原始 JSON 字典。

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 會處理回應： SDK 說明：
  • 檢查 HTTP 狀態碼。
  • 解析回應主體（通常是 JSON）。
  • 驗證資料是否符合預期的結構。
  • 將資料映射到 Python 模型物件。
• 你的程式碼會接收 Python 物件： 你處理的是反序列化的物件，而不是原始的 HTTP 資料。

此抽象可讓您專注於應用程式邏輯，而不是 HTTP 通訊協定詳細資料。

常見的回應類型

適用於 Python 的 Azure SDK 會在所有服務中使用數個標準回應類型。 瞭解這些類型可協助您有效地使用任何 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 範本。
• 定型機器學習模型。
• 複製大型二進位物件。

這些作業會傳回追蹤作業進度的輪詢器物件。

與民調人員的工作

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