Informazioni sui tipi di risposta comuni in Azure SDK per Python

Azure SDK per Python astrae le chiamate al protocollo di comunicazione del servizio di Azure sottostante, indipendentemente dal fatto che il protocollo sia HTTP o AMQP (usato per gli SDK di messaggistica come ServiceBus e EventHubs). Ad esempio, se si usa una delle librerie che usano HTTP, Azure SDK per Python effettua richieste HTTP e riceve risposte HTTP sotto le quinte. L'SDK astrae questa complessità in modo che sia possibile usare oggetti Python intuitivi invece di risposte HTTP non elaborate o payload JSON.

Comprendere i tipi di oggetti ricevuti dalle operazioni dell'SDK è essenziale per la scrittura di applicazioni Azure efficaci. Questo articolo illustra i tipi di risposta comuni riscontrati e il modo in cui sono correlati alla comunicazione HTTP sottostante.

Annotazioni

Questo articolo esamina solo lo scenario HTTP, non lo scenario AMQP.

Oggetti Python deserializzati

Azure SDK per Python assegna priorità alla produttività degli sviluppatori restituendo oggetti Python fortemente tipizzati dalle operazioni del servizio. Anziché analizzare JSON o gestire direttamente i codici di stato HTTP, è possibile usare modelli di risorse che rappresentano le risorse di Azure come oggetti Python.

Ad esempio, quando si recupera un BLOB da Archiviazione di Azure, si riceve un BlobProperties oggetto con attributi come name, sizee last_modified anziché un dizionario JSON non elaborato:

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

Da dove provengono i dati

Comprendere il flusso di dati consente di apprezzare le operazioni dell'SDK in background:

• Il codice chiama un metodo SDK: Si richiama un metodo come get_blob_properties().
• L'SDK costruisce una richiesta HTTP: L'SDK compila la richiesta HTTP appropriata con intestazioni, autenticazione e parametri di query.
• Il servizio di Azure risponde: Il servizio restituisce una risposta HTTP, in genere con un payload JSON nel corpo della risposta.
• L'SDK elabora la risposta: The SDK:
  • Controlla il codice di stato HTTP.
  • Analizza il corpo della risposta (in genere JSON).
  • Convalida i dati rispetto agli schemi previsti.
  • Esegue il mapping dei dati agli oggetti modello Python.
• Il codice riceve oggetti Python: È possibile usare gli oggetti deserializzati, non i dati HTTP non elaborati.

Questa astrazione consente di concentrarsi sulla logica dell'applicazione anziché sui dettagli del protocollo HTTP.

Tipi di risposta comuni

Azure SDK per Python usa diversi tipi di risposta standard in tutti i servizi. La comprensione di questi tipi consente di lavorare in modo efficace con qualsiasi servizio di Azure.

Modelli di risorse

La maggior parte delle operazioni SDK restituisce modelli di risorse. Questi oggetti Python rappresentano le risorse di Azure. I modelli sono specifici del servizio, ma seguono modelli coerenti:

# 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 per i risultati della raccolta

Quando l'SDK elenca le risorse, restituisce ItemPaged oggetti che gestiscono la paginazione in modo trasparente:

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

Accedere alla risposta HTTP non elaborata

Anche se le astrazioni di alto livello dell'SDK soddisfano la maggior parte delle esigenze, a volte è necessario accedere alla risposta HTTP sottostante. Gli scenari comuni includono:

• Debug di richieste non riuscite.
• Accesso alle intestazioni di risposta personalizzate.
• Implementazione della logica di ripetizione dei tentativi personalizzata.
• Utilizzo dei formati di risposta non standard.

La maggior parte dei metodi SDK accetta un raw_response_hook parametro:

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)

Paging e iteratori

I servizi di Azure spesso restituiscono grandi raccolte di risorse. L'SDK usa ItemPaged per gestire queste raccolte in modo efficiente senza caricare tutto in memoria contemporaneamente.

Paginazione automatica

L'SDK recupera automaticamente nuove pagine durante l'iterazione:

# 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

Lavorare con le pagine in modo esplicito

È anche possibile usare le pagine direttamente quando necessario:

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)

Controlla la dimensione della pagina

Molte operazioni di elenco accettano un results_per_page parametro:

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

Alcuni metodi per alcuni servizi di Azure hanno altri meccanismi per controllare le dimensioni della pagina. Ad esempio, Azure Key Vault e Ricerca di Azure usano il top kwarg per limitare i risultati per ogni chiamata. Per un esempio che usa il metodo Ricerca di search() Azure, vedere il codice sorgente.

Caso speciale: operazioni e poller a esecuzione prolungata

Alcune operazioni di Azure non possono essere completate immediatamente. Gli esempi includono:

• Creazione o eliminazione di macchine virtuali.
• Distribuzione di modelli di Azure Resource Manager.
• Addestrare modelli di apprendimento automatico.
• Copiare BLOB di grandi dimensioni.

Queste operazioni restituiscono oggetti poller che tengono traccia dello stato dell'operazione.

Collaborare con i poller

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

Poller asincroni

Quando si usano modelli asincroni/await, si lavora con 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()

Esempio di polling di oggetti per operazioni a esecuzione prolungata: Macchine virtuali

La distribuzione di macchine virtuali è un esempio di un'operazione che richiede tempo per completare e gestiscla restituendo oggetti poller (LROPoller per il codice sincrono, AsyncLROPoller per il codice asincrono):

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

Accedere alla risposta per i risultati di paging

Per i risultati di paging, usare il by_page() metodo con 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)

Procedure consigliate

• Preferisce astrazioni di alto livello.

• Usare i modelli di risorse dell'SDK anziché le risposte non elaborate quando possibile.

• Evitare di accedere a qualsiasi metodo preceduto da un carattere di sottolineatura (_). Per convenzione, questi metodi sono privati in Python. Non esistono garanzie relative a problemi come modifiche che rompono la compatibilità rispetto alle API pubbliche.

  # 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'])
  

• Gestire correttamente la paginazione. Scorrere i risultati paginati anziché convertirli sempre in un elenco.

  # 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
  

• Usare poller.result() per le operazioni a esecuzione prolungata. Usare sempre il result() metodo per assicurarsi che le operazioni vengano completate correttamente:

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

• Accedere alle risposte non elaborate solo quando necessario. Usare con moderazione l'accesso alle risposte non elaborate e solo per requisiti specifici:

  # 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