Compreender os tipos de resposta comuns no Azure SDK para Python

O Azure SDK para Python abstrai chamadas para o protocolo de comunicação de serviços Azure subjacente, quer esse protocolo seja HTTP ou AMQP (que é usado para SDKs de mensagens como ServiceBus e EventHubs). Por exemplo, se usar uma das bibliotecas que utiliza HTTP, o Azure SDK para Python faz pedidos HTTP e recebe respostas HTTP em segundo plano. O SDK abstrai esta complexidade para que possas trabalhar com objetos Python intuitivos em vez de respostas HTTP brutas ou cargas úteis JSON.

Compreender os tipos de objetos que recebe das operações do SDK é essencial para escrever aplicações Azure eficazes. Este artigo explica os tipos de resposta mais comuns que encontra e como se relacionam com a comunicação HTTP subjacente.

Observação

Este artigo examina apenas o cenário HTTP, não o cenário AMQP.

Objetos Python desserializados

O SDK do Azure para Python prioriza a produtividade do desenvolvedor retornando objetos Python fortemente tipados de operações de serviço. Em vez de analisar JSON ou manipular códigos de status HTTP diretamente, você trabalha com modelos de recursos que representam recursos do Azure como objetos Python.

Por exemplo, quando recupera um blob do Azure Storage, recebe um BlobProperties objeto com atributos como name, size, e last_modified em vez de um dicionário JSON bruto:

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

De onde provêm os dados

Compreender o fluxo de dados ajuda você a apreciar o que o SDK faz nos bastidores:

• O seu código chama um método SDK: Invocas um método como get_blob_properties().
• O SDK constrói um pedido HTTP: O SDK constrói o pedido HTTP apropriado com cabeçalhos, autenticação e parâmetros de consulta.
• O serviço Azure responde: O serviço devolve uma resposta HTTP, normalmente com uma carga útil JSON no corpo da resposta.
• O SDK processa a resposta: O SDK:
  • Verifica o código de estado HTTP.
  • Analisa o corpo da resposta (normalmente JSON).
  • Valida os dados em relação aos esquemas esperados.
  • Mapeia os dados para objetos do modelo Python.
• O seu código recebe objetos Python: Trabalhas com os objetos deserializados, não com dados HTTP brutos.

Essa abstração permite que você se concentre na lógica do aplicativo em vez dos detalhes do protocolo HTTP.

Tipos de resposta comuns

O SDK do Azure para Python usa vários tipos de resposta padrão em todos os serviços. Compreender estes tipos ajuda-o a trabalhar de forma eficaz com qualquer serviço do Azure.

Modelos de recursos

A maioria das operações SDK devolve modelos de recursos. Estes objetos Python representam recursos Azure. Os modelos são específicos de serviço, mas seguem padrões consistentes:

# 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 para resultados da coleção

Quando o SDK lista recursos, devolve ItemPaged objetos que tratam a paginação de forma transparente:

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

Aceda à resposta HTTP bruta

Embora as abstrações de alto nível do SDK atendam à maioria das necessidades, às vezes você precisa acessar a resposta HTTP subjacente. Cenários comuns incluem:

• Depuração de requerimentos falhados.
• Aceder a cabeçalhos de resposta personalizados.
• Implementar lógica de tentativas personalizada.
• Trabalhar com formatos de resposta não padrão.

A maioria dos métodos SDK aceita um raw_response_hook parâmetro:

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)

Paginação e iteradores

Os serviços do Azure geralmente retornam grandes coleções de recursos. O SDK utiliza ItemPaged para gerir estas coleções de forma eficiente sem carregar tudo na memória de uma só vez.

Paginação automática

O SDK busca automaticamente novas páginas à medida que você itera:

# 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

Trabalhar com páginas de forma explícita

Você também pode trabalhar com páginas diretamente quando necessário:

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)

Controlar o tamanho da página

Muitas operações de lista aceitam um results_per_page parâmetro:

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

Alguns métodos para alguns serviços do Azure têm outros mecanismos para controlar o tamanho da página. Por exemplo, o Azure Key Vault e o Azure Search usam o top kwarg para limitar resultados por chamada. Para um exemplo que utiliza o método Azure Search search() , veja o código-fonte.

Caso especial: Operações de longa duração e sondagens

Algumas operações do Azure não podem ser concluídas imediatamente. Os exemplos incluem:

• Criar ou eliminar máquinas virtuais.
• Implementação de modelos do Azure Resource Manager.
• Treinamento de modelos de aprendizado de máquina.
• Copiar grandes blobs.

Essas operações retornam objetos poller que rastreiam o progresso da operação.

Trabalho com os interrogadores

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

Sondadores assíncronos

Quando usas padrões async/await, trabalhas com 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()

Exemplo de sondagem de objetos para operações de longa duração: máquinas virtuais

A implementação de máquinas virtuais é um exemplo de uma operação que demora tempo a concluir e é gerida retornando objetos poller (LROPoller para código síncrono, AsyncLROPoller para código assíncrono).

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

Resposta de acesso para resultados paginados

Para resultados paginados, use o by_page() método com 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)

Melhores práticas

• Prefiro abstrações de alto nível.

• Trabalhe com os modelos de recursos do SDK em vez de respostas brutas sempre que possível.

• Evite aceder a qualquer método com o prefixo sublinhado (_). Por convenção, esses métodos são privados em Python. Não há garantias quanto a problemas como alterações irregulares em comparação com APIs públicas:

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

• Manuseie a paginação corretamente. Itere sempre sobre resultados paginados em vez de converter para uma lista:

  # 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
  

• Use poller.result() para operações de longa duração. Use sempre o result() método para garantir que as operações são concluídas com sucesso:

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

• Aceda às respostas brutas apenas quando necessário. Use o acesso de resposta bruta com moderação e apenas para requisitos específicos:

  # 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