Entender os tipos de resposta comuns no SDK do Azure para Python

O SDK do Azure para Python abstrai chamadas para o protocolo de comunicação de serviço subjacente do Azure, seja ele HTTP ou AMQP (que é usado para SDKs de mensagens como ServiceBus e EventHubs). Por exemplo, se você usar uma das bibliotecas que usa HTTP, o SDK do Azure para Python fará solicitações HTTP e receberá respostas HTTP nos bastidores. O SDK abstrai essa complexidade para que você possa trabalhar com objetos Python intuitivos em vez de respostas HTTP brutas ou cargas JSON.

Entender os tipos de objetos recebidos das operações do SDK é essencial para escrever aplicativos eficazes do Azure. Este artigo explica os tipos de resposta comuns que você encontra e como eles 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 tipado de operações de serviço. Em vez de analisar o JSON ou lidar diretamente com códigos de status HTTP, você trabalha com modelos de recursos que representam recursos do Azure como objetos Python.

Por exemplo, quando você recupera um blob do Armazenamento do Azure, recebe um BlobProperties objeto com atributos como name, sizee 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 os dados vêm

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

  • Seu código chama um método SDK: Você invoca um método como get_blob_properties().
  • O SDK constrói uma solicitação HTTP: O SDK cria a solicitação HTTP apropriada com cabeçalhos, autenticação e parâmetros de consulta.
  • O serviço do Azure responde: O serviço retorna uma resposta HTTP, normalmente com uma carga JSON no corpo da resposta.
  • O SDK processa a resposta: O SDK:
    • Verifica o código de status HTTP.
    • Analisa o corpo da resposta (geralmente JSON).
    • Valida os dados em relação aos esquemas esperados.
    • Mapeia os dados para objetos de modelo do Python.
  • Seu código recebe objetos Python: Você trabalha com os objetos desserializados, não com dados HTTP brutos.

Essa abstração permite que você se concentre na lógica do aplicativo em vez de nos 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. Entender esses tipos ajuda você a trabalhar efetivamente com qualquer serviço do Azure.

Modelos de recursos

A maioria das operações do SDK retorna modelos de recurso. Esses objetos Python representam recursos do Azure. Os modelos são específicos do 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, ele retorna ItemPaged objetos que manipulam 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}")

Acessar a resposta HTTP bruta

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

  • Depuração de solicitações falhadas.
  • Acessando cabeçalhos de resposta personalizados.
  • Implementando uma lógica de tentativa de nova execução personalizada.
  • Trabalhando com formatos de resposta não padrão.

A maioria dos métodos do 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 usa ItemPaged para lidar com essas coleções com eficiência sem carregar tudo na memória ao mesmo tempo.

Paginação automática

O SDK busca automaticamente novas páginas conforme 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 explicitamente

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)

Tamanho da página de controle

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 os resultados por chamada. Para obter um exemplo que usa o método do Azure Search search() , consulte o código-fonte.

Caso especial: operações de execução longa e sondadores

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

  • Criando ou excluindo máquinas virtuais.
  • Implantando modelos do Azure Resource Manager.
  • Treinamento de modelos de aprendizado de máquina.
  • Copiando blobs grandes.

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

Trabalhar com sondadores

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

Ao usar padrões assíncronos/de espera, você trabalha 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 objetos de sondagem para operações de execução longa: máquinas virtuais

A implantação de máquinas virtuais é um exemplo de uma operação que leva tempo para ser concluída e manipula-a 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 paginado, 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)

Práticas recomendadas

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

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

  • Evite acessar qualquer método prefixado com um sublinhado (_). Por convenção, esses métodos são privados no Python. Não há garantias sobre questões como alterações de ruptura, quando comparadas a 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'])

  • Manipule a paginação corretamente. Sempre iterar sobre os resultados paginados em vez de converter em 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. Sempre use o result() método para garantir que as operações sejam concluídas com êxito:

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

  • Acesse respostas brutas somente quando necessário. Use o acesso de resposta bruta com moderação e somente 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
  • Last updated on