Comprendre les types de réponse courants dans le Kit de développement logiciel (SDK) Azure pour Python

Le Kit de développement logiciel (SDK) Azure pour Python abstrait les appels au protocole de communication du service Azure sous-jacent, que ce protocole soit HTTP ou AMQP (qui est utilisé pour les SDKs de messagerie tels que ServiceBus et EventHubs). Par exemple, si vous utilisez l’une des bibliothèques qui utilisent HTTP, le Kit de développement logiciel (SDK) Azure pour Python effectue des requêtes HTTP et reçoit des réponses HTTP sous le capot. Le Kit de développement logiciel (SDK) extrait cette complexité pour que vous puissiez utiliser des objets Python intuitifs au lieu de réponses HTTP brutes ou de charges utiles JSON.

Comprendre les types d’objets que vous recevez à partir des opérations du SDK est essentiel pour écrire des applications Azure efficaces. Cet article explique les types de réponse courants que vous rencontrez et leur relation avec la communication HTTP sous-jacente.

Note

Cet article examine uniquement le scénario HTTP, et non le scénario AMQP.

Objets Python désérialisés

Le Kit de développement logiciel (SDK) Azure pour Python hiérarchise la productivité des développeurs en retournant des objets Python fortement typés à partir d’opérations de service. Au lieu d’analyser des codes d’état JSON ou de gérer directement les codes d’état HTTP, vous travaillez avec des modèles de ressources qui représentent des ressources Azure en tant qu’objets Python.

Par exemple, lorsque vous récupérez un objet blob à partir du Stockage Azure, vous recevez un BlobProperties objet avec des attributs tels que name, sizeet last_modified plutôt qu’un dictionnaire JSON brut :

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

Où proviennent les données

La compréhension du flux de données vous permet d’apprécier ce que fait le Kit de développement logiciel (SDK) en arrière-plan :

Votre code appelle une méthode sdk : Vous appelez une méthode comme get_blob_properties().
Le Kit de développement logiciel (SDK) construit une requête HTTP : Le Kit de développement logiciel (SDK) génère la requête HTTP appropriée avec les en-têtes, l’authentification et les paramètres de requête.
Le service Azure répond : Le service retourne une réponse HTTP, généralement avec une charge utile JSON dans le corps de la réponse.
Le Kit de développement logiciel (SDK) traite la réponse : Kit de développement logiciel (SDK) :
- Vérifie le code d’état HTTP.
- Analyse le corps de la réponse (généralement JSON).
- Valide les données par rapport aux schémas attendus.
- Mappe les données aux objets de modèle Python.
Votre code reçoit des objets Python : Vous utilisez les objets désérialisés, et non les données HTTP brutes.

Cette abstraction vous permet de vous concentrer sur votre logique d’application plutôt que sur les détails du protocole HTTP.

Types de réponse courants

Le Kit de développement logiciel (SDK) Azure pour Python utilise plusieurs types de réponse standard dans tous les services. La compréhension de ces types vous aide à travailler efficacement avec n’importe quel service Azure.

Modèles de ressources

La plupart des opérations du SDK retournent des modèles de ressources. Ces objets Python représentent des ressources Azure. Les modèles sont spécifiques au service, mais suivent des modèles cohérents :

# 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 pour les résultats de la collection

Lorsque le Kit de développement logiciel (SDK) répertorie les ressources, il retourne ItemPaged des objets qui gèrent la pagination de manière 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}")

Accéder à la réponse HTTP brute

Bien que les abstractions de haut niveau du Kit de développement logiciel (SDK) répondent à la plupart des besoins, vous avez parfois besoin d’accéder à la réponse HTTP sous-jacente. Les scénarios courants sont les suivants :

Débogage des requêtes échouées.
Accès aux en-têtes de réponse personnalisés.
Implémentation d’une logique de nouvelle tentative personnalisée.
Utilisation de formats de réponse non standard.

La plupart des méthodes sdk acceptent un raw_response_hook paramètre :

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)

Pagination et itérateurs

Les services Azure retournent souvent de grandes collections de ressources. Le Kit de développement logiciel (SDK) utilise ItemPaged pour gérer efficacement ces collections sans tout charger en mémoire en même temps.

Pagination automatique

Le Kit de développement logiciel (SDK) récupère automatiquement les nouvelles pages lors de l’itération :

# 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

Travailler avec des pages explicitement

Vous pouvez également utiliser des pages directement si nécessaire :

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)

Taille de la page de contrôle

De nombreuses opérations de liste acceptent un results_per_page paramètre :

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

Certaines méthodes pour certains services Azure ont d’autres mécanismes pour contrôler la taille de page. Par exemple, Azure Key Vault et Recherche Azure utilisent le top kwarg pour limiter les résultats par appel. Pour obtenir un exemple qui utilise la méthode Recherche search() Azure, consultez le code source.

Cas spécial : opérations et sondages de longue durée

Certaines opérations Azure ne peuvent pas se terminer immédiatement. Voici quelques exemples :

Création ou suppression de machines virtuelles.
Déploiement de modèles Azure Resource Manager.
Entraînement de modèles Machine Learning.
Copie de BLOBs volumineux.

Ces opérations retournent des objets d’polleur qui suivent la progression de l’opération.

Utiliser des interrogateurs

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

Polleurs asynchrones

Lorsque vous utilisez des modèles asynchrones/await, vous utilisez 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()

Exemple d'interrogation d'objets pour des opérations de longue durée : les machines virtuelles

Le déploiement de machines virtuelles est un exemple d’opération qui nécessite du temps pour s’achever, et cette gestion se fait en retournant des objets d’interrogation (LROPoller pour le code synchrone, AsyncLROPoller pour le code asynchrone) :

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

Réponse d’accès pour les résultats paginés

Pour les résultats paginés, utilisez la by_page() méthode avec 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)

Meilleures pratiques

Préférer les abstractions de haut niveau.
Utilisez les modèles de ressources du Kit de développement logiciel (SDK) plutôt que les réponses brutes dans la mesure du possible.

Évitez d’accéder à une méthode précédée d’un trait de soulignement (_). Par convention, ces méthodes sont privées dans Python. Il n’existe aucune garantie concernant les problèmes tels que les modifications perturbatrices des API publiques :

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

Gérez correctement la pagination. Effectuez toujours une itération sur les résultats paginés au lieu de convertir en liste :

# 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

Utilisez poller.result() pour les opérations de longue durée. Utilisez toujours la result() méthode pour vous assurer que les opérations se terminent correctement :

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

Accéder aux réponses brutes uniquement si nécessaire. Utilisez l’accès de réponse brute avec parcimonie et uniquement pour des exigences spécifiques :

# 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

Commentaires

Cette page a-t-elle été utile ?

Dernière mise à jour : 2025-12-09