Compartir a través de

Descripción de los tipos de respuesta comunes en el SDK de Azure para Python

El SDK de Azure para Python abstrae las llamadas al protocolo de comunicación del servicio de Azure subyacente, tanto si ese protocolo es HTTP como AMQP (que se usa para los SDK de mensajería como ServiceBus y EventHubs). Por ejemplo, si usa una de las bibliotecas que usa HTTP, el SDK de Azure para Python realiza solicitudes HTTP y recibe respuestas HTTP en segundo plano. El SDK abstrae esta complejidad para que pueda trabajar con objetos intuitivos de Python en lugar de respuestas HTTP sin procesar o cargas JSON.

Comprender los tipos de objetos que recibe de las operaciones del SDK es esencial para escribir aplicaciones eficaces de Azure. En este artículo se explican los tipos de respuesta comunes que se encuentran y cómo se relacionan con la comunicación HTTP subyacente.

Nota:

En este artículo solo se examina el escenario HTTP, no el escenario AMQP.

Objetos de Python deserializados

El SDK de Azure para Python prioriza la productividad del desarrollador devolviendo objetos de Python fuertemente tipados a partir de operaciones de servicio. En lugar de analizar JSON o controlar directamente los códigos de estado HTTP, se trabaja con modelos de recursos que representan recursos de Azure como objetos de Python.

Por ejemplo, al recuperar un blob de Azure Storage, recibirá un BlobProperties objeto con atributos como name, sizey last_modified en lugar de un diccionario JSON sin formato:

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

Dónde proceden los datos

Comprender el flujo de datos le ayuda a apreciar lo que hace el SDK en segundo plano:

  • El código llama a un método del SDK: Se invoca un método como get_blob_properties().
  • El SDK crea una solicitud HTTP: El SDK compila la solicitud HTTP adecuada con encabezados, autenticación y parámetros de consulta.
  • El servicio de Azure responde: El servicio devuelve una respuesta HTTP, normalmente con una carga JSON en el cuerpo de la respuesta.
  • El SDK procesa la respuesta: El SDK:
    • Comprueba el código de estado HTTP.
    • Analiza el cuerpo de la respuesta (normalmente JSON).
    • Valida los datos con los esquemas esperados.
    • Asigna los datos a objetos de modelo de Python.
  • El código recibe objetos de Python: Trabaja con los objetos deserializados, no con datos HTTP sin procesar.

Esta abstracción le permite centrarse en la lógica de la aplicación en lugar de en los detalles del protocolo HTTP.

Tipos de respuesta comunes

El SDK de Azure para Python usa varios tipos de respuesta estándar en todos los servicios. Comprender estos tipos le ayuda a trabajar de forma eficaz con cualquier servicio de Azure.

Modelos de recursos

La mayoría de las operaciones del SDK devuelven modelos de recursos. Estos objetos de Python representan recursos de Azure. Los modelos son específicos del servicio, pero siguen patrones coherentes:

# 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 los resultados de la colección

Cuando el SDK enumera los recursos, devuelve ItemPaged objetos que controlan la paginación 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}")

Acceso a la respuesta HTTP sin procesar

Aunque las abstracciones de alto nivel del SDK satisfacen la mayoría de las necesidades, a veces necesita acceso a la respuesta HTTP subyacente. Entre los escenarios habituales se incluyen los siguientes:

  • Depuración de solicitudes fallidas.
  • Acceso a encabezados de respuesta personalizados.
  • Implementación de la lógica de reintento personalizada.
  • Trabajar con formatos de respuesta no estándar.

La mayoría de los métodos del SDK aceptan un 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)

Paginación e iteradores

Los servicios de Azure suelen devolver grandes colecciones de recursos. El SDK usa ItemPaged para controlar estas colecciones de forma eficaz sin cargar todo en memoria a la vez.

Paginación automática

El SDK captura automáticamente las páginas nuevas a medida que recorre en iteración:

# 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

Trabajar explícitamente con páginas

También puede trabajar con páginas directamente cuando sea necesario:

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)

Control del tamaño de página

Muchas operaciones de lista aceptan un results_per_page parámetro:

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

Algunos métodos para algunos servicios de Azure tienen otros mecanismos para controlar el tamaño de página. Por ejemplo, Azure Key Vault y Azure Search usan top kwarg para limitar los resultados por llamada. Para ver un ejemplo que usa el método Azure Search search() , consulte el código fuente.

Caso especial: operaciones y sondeos de larga duración

Algunas operaciones de Azure no se pueden completar inmediatamente. Algunos ejemplos son:

  • Creación o eliminación de máquinas virtuales.
  • Implementación de plantillas de Azure Resource Manager.
  • Entrenamiento de modelos de aprendizaje automático.
  • Copiando blobs grandes.

Estas operaciones devuelven objetos de sondeo que realizan un seguimiento del progreso de la operación.

Trabajar con sondeos

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

Sondeos asincrónicos

Cuando se utilizan patrones async/await, se trabaja 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()

Ejemplo de sondeo de objetos para operaciones de larga duración: Máquinas virtuales

La implementación de máquinas virtuales es un ejemplo de una operación que tarda tiempo en completarse y se gestiona devolviendo objetos de encuesta (LROPoller para código sincrónico, AsyncLROPoller para código asincrónico):

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

Respuesta de acceso a los resultados paginados

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

procedimientos recomendados

  • Prefiere abstracciones de alto nivel.

  • Trabaje con los modelos de recursos del SDK en lugar de las respuestas sin procesar siempre que sea posible.

  • Evite tener acceso a cualquier método con prefijo con un carácter de subrayado (_). Por convención, esos métodos son privados en Python. No hay garantías sobre problemas como cambios importantes en comparación con las API 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'])

  • Controle correctamente la paginación. Siempre recorre en iteración los resultados paginados en lugar de convertir en una 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 las operaciones de ejecución prolongada. Use siempre el result() método para asegurarse de que las operaciones se completen correctamente:

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

  • Acceda a las respuestas sin procesar solo cuando sea necesario. Utiliza el acceso de respuesta en bruto con moderación y solo 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