Vysvětlení běžných typů odpovědí v sadě Azure SDK pro Python

Sada Azure SDK pro Python abstrahuje volání základního komunikačního protokolu služby Azure, ať už jde o protokol HTTP nebo AMQP (který se používá pro sady SDK pro zasílání zpráv, jako a ServiceBusEventHubs). Pokud například používáte některou z knihoven, která používá PROTOKOL HTTP, azure SDK pro Python odesílá požadavky HTTP a přijímá odpovědi HTTP pod kapotou. Sada SDK tuto složitost abstrahuje, abyste mohli pracovat s intuitivními objekty Pythonu místo nezpracovaných odpovědí HTTP nebo datových částí JSON.

Pochopení typů objektů, které získáváte z operací sady SDK, je nezbytné pro psaní efektivních aplikací Azure. Tento článek vysvětluje běžné typy odpovědí, se kterými se setkáte, a jak souvisí se základní komunikací HTTP.

Deserializované objekty Pythonu

Sada Azure SDK pro Python upřednostňuje produktivitu vývojářů tím, že z operací služby vrací silně typované objekty Pythonu. Místo přímé analýzy JSON nebo zpracování stavových kódů HTTP pracujete s modely prostředků, které představují prostředky Azure jako objekty Pythonu.

Například když načtete objekt blob ze služby Azure Storage, obdržíte BlobProperties objekt s atributy, jako name, size a last_modified, nikoli nezpracovaný slovník 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}")

Odkud data pocházejí

Pochopení toku dat vám pomůže pochopit, co sada SDK na pozadí dělá:

• Váš kód volá metodu sady SDK: Vyvoláte metodu jako get_blob_properties().
• Sada SDK vytvoří požadavek HTTP: Sada SDK sestaví odpovídající požadavek HTTP s hlavičkami, ověřováním a parametry dotazu.
• Služba Azure reaguje: Služba vrátí odpověď HTTP, obvykle s datovou částí JSON v textu odpovědi.
• Sada SDK zpracuje odpověď: Sada SDK:
  • Zkontroluje stavový kód HTTP.
  • Parsuje text odpovědi (obvykle JSON).
  • Ověří data proti očekávaným schématům.
  • Mapuje data na objekty modelu Pythonu.
• Váš kód přijímá objekty Pythonu: Pracujete s deserializovanými objekty, nikoli s nezpracovanými daty HTTP.

Tato abstrakce umožňuje zaměřit se na logiku aplikace, nikoli na podrobnosti protokolu HTTP.

Běžné typy odpovědí

Sada Azure SDK pro Python používá ve všech službách několik standardních typů odpovědí. Porozumění těmto typům vám pomůže efektivně pracovat s libovolnou službou Azure.

Modely prostředků

Většina operací sady SDK vrací modely prostředků. Tyto objekty Pythonu představují prostředky Azure. Modely jsou specifické pro službu, ale dodržují konzistentní vzory:

# 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 pro výsledky kolekce

Když sada SDK vypíše prostředky, vrátí ItemPaged objekty, které zpracovávají stránkování transparentně:

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}")

Přístup k nezpracované odpovědi HTTP

I když abstrakce vysoké úrovně sady SDK splňují většinu potřeb, někdy potřebujete přístup k podkladové odpovědi HTTP. Mezi běžné scénáře patří:

• Ladění chybných požadavků
• Přístup k vlastním hlavičkám odpovědi
• Implementace vlastní logiky opakování
• Práce s nestandardními formáty odpovědí

Většina metod sady SDK přijímá raw_response_hook parametr:

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)

Stránkování a iterátory

Služby Azure často vracejí velké kolekce prostředků. Sada SDK používá ItemPaged k efektivnímu zpracování těchto kolekcí bez načtení všeho do paměti najednou.

Automatická paginace

Sada SDK při iteraci automaticky načte nové stránky:

# 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

Explicitní práce se stránkami

Můžete také pracovat se stránkami přímo v případě potřeby:

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)

Velikost stránky ovládacího prvku

Mnoho operací seznamu přijímá results_per_page parametr:

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

Některé metody některých služeb Azure mají jiné mechanismy pro řízení velikosti stránky. Například Azure Key Vault a Azure Search používají top kwarg k omezení výsledků na volání. Příklad, který používá metodu Azure Search search() , najdete ve zdrojovém kódu.

Zvláštní případ: Dlouhotrvající operace a pollery

Některé operace Azure se nedají dokončit okamžitě. Mezi příklady patří:

• Vytváření nebo odstraňování virtuálních počítačů
• Nasazení šablon Azure Resource Manageru
• Trénování modelů strojového učení
• Kopírování velkých blobů

Tyto operace vrací objekty typu poller, které se zaměřují na sledování průběhu operace.

Práce s pollery

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

Asynchronní dotazovací nástroje

Pokud používáte vzory async/await, pracujete s 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()

Objekty dotazování pro dlouhotrvající operace: příklad virtuální počítače

Nasazení virtuálních počítačů je příkladem operace, která nějakou dobu trvá dokončit a řešení spočívá ve vrácení přehledových objektů (LROPoller pro synchronní kód, AsyncLROPoller pro asynchronní kód):

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")

Odpověď na přístup pro stránkované výsledky

Pro stránkované výsledky použijte metodu by_page() s 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)

Osvědčené postupy

• Upřednostněte abstrakce vysoké úrovně.

• Pracujte s modely prostředků sady SDK místo nezpracovaných odpovědí, kdykoli je to možné.

• Vyhněte se přístupu k jakékoli metodě s předponou podtržítka (_). Podle konvence jsou tyto metody v Pythonu soukromé. Neexistují žádné záruky problémů, jako jsou zásadní změny ve srovnání s veřejnými rozhraními 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'])
  

• Správně zpracujte stránkování. Vždy iterujte stránkované výsledky místo převodu na seznam:

  # 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
  

• Pro dlouhotrvající operace použijte příkaz poller.result(). Vždy použijte metodu result() k zajištění úspěšného dokončení operací.

  # 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!
  

• Přístup k nezpracovaných odpovědím pouze v případě potřeby Nezpracovaný přístup k odpovědím používejte střídmě a pouze pro konkrétní požadavky:

  # 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