Gérer les erreurs produites par le Kit de développement logiciel (SDK) Azure pour Python

La création d’applications cloud fiables nécessite plus que l’implémentation de fonctionnalités. Elle exige également des stratégies robustes de gestion des erreurs. Lorsque vous travaillez avec des systèmes distribués et des services cloud, votre application doit être prête à gérer correctement les scénarios d’échec.

Le Kit de développement logiciel (SDK) Azure pour Python fournit un modèle d’erreur complet conçu pour aider les développeurs à créer des applications résilientes. Comprendre ce modèle d’erreur est essentiel pour :

• Amélioration de la fiabilité des applications en anticipant et en gérant les scénarios d’échec courants.
• Amélioration de l’expérience utilisateur grâce à des messages d’erreur significatifs et à une dégradation normale.
• Simplification de la résolution des problèmes en capturant et en journalisant les informations de diagnostic pertinentes.

Cet article explore l’architecture des erreurs du Kit de développement logiciel (SDK) Azure pour Python et fournit des conseils pratiques pour implémenter une gestion efficace des erreurs dans vos applications.

Comment les erreurs du Kit de développement logiciel (SDK) Azure pour Python sont-ils ?

Le Kit de développement logiciel (SDK) Azure pour Python utilise un modèle d’exception hiérarchique qui fournit des fonctionnalités générales et spécifiques de gestion des erreurs. Au cœur de ce modèle est AzureError, qui sert de classe d’exception de base pour toutes les erreurs liées au Kit de développement logiciel (SDK) Azure.

Hiérarchie d’exceptions

AzureError
├── ClientAuthenticationError
├── ResourceNotFoundError
├── ResourceExistsError
├── ResourceModifiedError
├── ResourceNotModifiedError
├── ServiceRequestError
├── ServiceResponseError
└── HttpResponseError

Types d’exceptions clés

Scénarios d’erreur courants

Comprendre les scénarios d’erreur classiques vous aide à implémenter des stratégies de gestion appropriées pour chaque situation.

Erreurs d’authentification et d’autorisation

Les échecs d’authentification se produisent lorsque le Kit de développement logiciel (SDK) ne peut pas vérifier votre identité :

from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

try:
    credential = DefaultAzureCredential()
    blob_service = BlobServiceClient(
        account_url="https://myaccount.blob.core.windows.net",
        credential=credential
    )
    # Attempt to list containers
    containers = blob_service.list_containers()
except ClientAuthenticationError as e:
    print(f"Authentication failed: {e.message}")
    # Don't retry - fix credentials first

Les erreurs d’autorisation (généralement HttpResponseError avec 403 état) se produisent lorsque vous n’avez pas d’autorisations :

from azure.core.exceptions import HttpResponseError

try:
    blob_client.upload_blob(data)
except HttpResponseError as e:
    if e.status_code == 403:
        print("Access denied. Check your permissions.")
    else:
        raise

Erreurs de ressource

Gérez correctement les ressources manquantes :

from azure.core.exceptions import ResourceNotFoundError

try:
    blob_client = container_client.get_blob_client("myblob.txt")
    content = blob_client.download_blob().readall()
except ResourceNotFoundError:
    print("Blob not found. Using default content.")
    content = b"default"

Empêcher la création de ressources en double :

from azure.core.exceptions import ResourceExistsError

try:
    container_client.create_container()
except ResourceExistsError:
    print("Container already exists.")
    # Continue with existing container

Erreurs de serveur

Gérez les défaillances côté serveur de manière appropriée :

from azure.core.exceptions import HttpResponseError

try:
    result = client.process_data(large_dataset)
except HttpResponseError as e:
    if 500 <= e.status_code < 600:
        print(f"Server error ({e.status_code}). The service may be temporarily unavailable.")
        # Consider retry logic here
    else:
        raise

Meilleures pratiques pour la gestion des erreurs

• Utilisez une gestion des exceptions spécifique : Interceptez toujours des exceptions spécifiques avant de revenir aux exceptions générales :

  from azure.core.exceptions import (
      AzureError,
      ClientAuthenticationError,
      ResourceNotFoundError,
      HttpResponseError
  )
  
  try:
      # Azure SDK operation
      result = client.get_resource()
  except ClientAuthenticationError:
      # Handle authentication issues
      print("Please check your credentials")
  except ResourceNotFoundError:
      # Handle missing resources
      print("Resource not found")
  except HttpResponseError as e:
      # Handle specific HTTP errors
      if e.status_code == 429:
          print("Rate limited. Please retry later.")
      else:
          print(f"HTTP error {e.status_code}: {e.message}")
  except AzureError as e:
      # Catch-all for other Azure errors
      print(f"Azure operation failed: {e}")
  

• Implémentez les stratégies de nouvelle tentative appropriées : Certaines erreurs justifient des tentatives de nouvelle tentative, tandis que d’autres ne le font pas.

  N’effectuez pas de nouvelle tentative :

  • 401 Non autorisé (échecs d’authentification)
  • 403 Interdit (échecs d’autorisation)
  • 400 Demande incorrecte (erreurs du client)
  • 404 Introuvable (sauf si vous attendez que la ressource apparaisse)

  Envisagez de réessayer sur :

  • 408 Délai d’expiration de la demande
  • 429 Trop de requêtes (avec interruption appropriée)
  • Erreur de serveur interne 500
  • Passerelle incorrecte 502
  • 503 Service indisponible
  • Délai d’expiration de la passerelle 504

• Extraire des informations d’erreur significatives

  from azure.core.exceptions import HttpResponseError
  
  try:
      client.perform_operation()
  except HttpResponseError as e:
      # Extract detailed error information
      print(f"Status code: {e.status_code}")
      print(f"Error message: {e.message}")
      print(f"Error code: {e.error.code if e.error else 'N/A'}")
  
      # Request ID is crucial for Azure support
      if hasattr(e, 'response') and e.response:
          request_id = e.response.headers.get('x-ms-request-id')
          print(f"Request ID: {request_id}")
  

Nouvelles tentatives de stratégies et résilience

Le Kit de développement logiciel (SDK) Azure inclut des mécanismes de nouvelle tentative intégrés qui gèrent automatiquement les défaillances temporaires.

Comportement de nouvelle tentative par défaut

La plupart des clients du Kit de développement logiciel (SDK) Azure incluent des stratégies de nouvelle tentative par défaut qui :

• Réessayez sur les erreurs de connexion et les codes d’état HTTP spécifiques.
• Utilisez un retardement exponentiel avec variation aléatoire.
• Limitez le nombre de nouvelles tentatives.

Personnaliser les stratégies de nouvelle tentative

Si le comportement par défaut ne correspond pas à votre cas d’usage, vous pouvez personnaliser la stratégie de nouvelle tentative :

from azure.storage.blob import BlobServiceClient
from azure.core.pipeline.policies import RetryPolicy

# Create a custom retry policy
retry_policy = RetryPolicy(
    retry_total=5,  # Maximum retry attempts
    retry_backoff_factor=2,  # Exponential backoff factor
    retry_backoff_max=60,  # Maximum backoff time in seconds
    retry_on_status_codes=[408, 429, 500, 502, 503, 504]
)

# Apply to client
blob_service = BlobServiceClient(
    account_url="https://myaccount.blob.core.windows.net",
    credential=credential,
    retry_policy=retry_policy
)

Éviter de gérer les erreurs de délai d’attente et de réseau avec des boucles personnalisées

Essayez d’utiliser des reprises intégrées pour les erreurs de réseau et de délai d’attente avant d’implémenter votre propre logique.

from azure.core.exceptions import ServiceRequestError
import time

# Avoid this approach if possible
max_retries = 3
retry_count = 0

while retry_count < max_retries:
    try:
        response = client.get_secret("mysecret")
        break
    except ServiceRequestError as e:
        retry_count += 1
        if retry_count >= max_retries:
            raise
        print(f"Network error. Retrying... ({retry_count}/{max_retries})")
        time.sleep(2 ** retry_count)  # Exponential backoff

Implémenter des modèles de rupture de circuit

Pour les opérations critiques, envisagez d’implémenter des modèles disjoncteur :

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half-open
    
    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = 'half-open'
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = func(*args, **kwargs)
            if self.state == 'half-open':
                self.state = 'closed'
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = 'open'
            
            raise e

Comprendre les messages d’erreur et les codes

Les services Azure retournent des réponses d’erreur structurées qui fournissent des informations de débogage précieuses.

• Analyser les réponses d’erreur

  from azure.core.exceptions import HttpResponseError
  import json
  
  try:
      client.create_resource(resource_data)
  except HttpResponseError as e:
      # Many Azure services return JSON error details
      if e.response and e.response.text():
          try:
              error_detail = json.loads(e.response.text())
              print(f"Error code: {error_detail.get('error', {}).get('code')}")
              print(f"Error message: {error_detail.get('error', {}).get('message')}")
  
              # Some services provide additional details
              if 'details' in error_detail.get('error', {}):
                  for detail in error_detail['error']['details']:
                      print(f"  - {detail.get('code')}: {detail.get('message')}")
          except json.JSONDecodeError:
              print(f"Raw error: {e.response.text()}")
  

• Capturez les informations de diagnostic : Capturez toujours les informations de diagnostic clés pour la résolution des problèmes :

  import logging
  from azure.core.exceptions import AzureError
  
  logger = logging.getLogger(__name__)
  
  try:
      result = client.perform_operation()
  except AzureError as e:
      # Log comprehensive error information
      logger.error(
          "Azure operation failed",
          extra={
              'error_type': type(e).__name__,
              'error_message': str(e),
              'operation': 'perform_operation',
              'timestamp': datetime.utcnow().isoformat(),
              'request_id': getattr(e.response, 'headers', {}).get('x-ms-request-id') if hasattr(e, 'response') else None
          }
      )
      raise
  

• Journalisation et diagnostics : Activez la journalisation au niveau du Kit de développement logiciel (SDK) pour une résolution des problèmes détaillée :

  import logging
  import sys
  
  # Configure logging for Azure SDKs
  logging.basicConfig(level=logging.DEBUG)
  
  # Enable HTTP request/response logging
  logging.getLogger('azure.core.pipeline.policies.http_logging_policy').setLevel(logging.DEBUG)
  
  # For specific services
  logging.getLogger('azure.storage.blob').setLevel(logging.DEBUG)
  logging.getLogger('azure.identity').setLevel(logging.DEBUG)
  

  Pour plus d’informations sur la journalisation, consultez Configurer la journalisation dans les bibliothèques Azure pour Python.

• Utilisez le suivi réseau : Pour le débogage approfondi, activez le suivi au niveau du réseau :

  Important

  La journalisation HTTP peut inclure des informations sensibles telles que des clés de compte dans les en-têtes et d’autres informations d’identification. Veillez à protéger ces journaux pour éviter de compromettre la sécurité.

  from azure.storage.blob import BlobServiceClient
  
  # Enable network tracing
  blob_service = BlobServiceClient(
      account_url="https://myaccount.blob.core.windows.net",
      credential=credential,
      logging_enable=True,  # Enable logging
      logging_body=True     # Log request/response bodies (careful with sensitive data)
  )
  

Considérations spéciales relatives à la programmation asynchrone

Lorsque vous utilisez des clients asynchrones, la gestion des erreurs nécessite une attention particulière.

• Gestion des erreurs asynchrones de base

  import asyncio
  from azure.core.exceptions import AzureError
  
  async def get_secret_async(client, secret_name):
      try:
          secret = await client.get_secret(secret_name)
          return secret.value
      except ResourceNotFoundError:
          print(f"Secret '{secret_name}' not found")
          return None
      except AzureError as e:
          print(f"Error retrieving secret: {e}")
          raise
  

• Gérer les annulations

  async def long_running_operation(client):
      try:
          result = await client.start_long_operation()
          # Wait for completion
          final_result = await result.result()
          return final_result
      except asyncio.CancelledError:
          print("Operation cancelled")
          # Cleanup if necessary
          if hasattr(result, 'cancel'):
              await result.cancel()
          raise
      except AzureError as e:
          print(f"Operation failed: {e}")
          raise
  

• Gestion simultanée des erreurs

  async def process_multiple_resources(client, resource_ids):
      tasks = []
      for resource_id in resource_ids:
          task = client.get_resource(resource_id)
          tasks.append(task)
  
      results = []
      errors = []
  
      # Use gather with return_exceptions to handle partial failures
      outcomes = await asyncio.gather(*tasks, return_exceptions=True)
  
      for resource_id, outcome in zip(resource_ids, outcomes):
          if isinstance(outcome, Exception):
              errors.append((resource_id, outcome))
          else:
              results.append(outcome)
  
      # Process successful results and errors appropriately
      if errors:
          print(f"Failed to process {len(errors)} resources")
          for resource_id, error in errors:
              print(f"  - {resource_id}: {error}")
  
      return results
  

Résumé des meilleures pratiques

La gestion efficace des erreurs dans le Kit de développement logiciel (SDK) Azure pour les applications Python vous oblige à :

• Anticiper les défaillances : Les applications cloud doivent s’attendre à des défaillances partielles et les gérer correctement.
• Utilisez une gestion des exceptions spécifique : Interceptez des exceptions spécifiques comme ResourceNotFoundError et ClientAuthenticationError avant de revenir à la gestion générale AzureError .
• Implémentez une logique de nouvelle tentative intelligente : Utilisez des stratégies de nouvelle tentative intégrées ou personnalisez-les en fonction de vos besoins. N’oubliez pas que toutes les erreurs ne doivent pas déclencher de nouvelles tentatives.
• Capturez les informations de diagnostic : Consignez toujours les ID de demande, les codes d’erreur et les horodatages pour une résolution efficace des problèmes.
• Fournissez des retours significatifs aux utilisateurs : Transformez les erreurs techniques en messages conviviaux tout en conservant les détails techniques pour l'assistance technique.
• Scénarios d’erreur de test : Incluez la gestion des erreurs dans votre couverture de test pour vous assurer que votre application se comporte correctement dans des conditions d’échec.

Contenu connexe

• Passez en revue la référence du module d’exceptions Azure Core.
• Découvrez comment résoudre les problèmes d’authentification et d’autorisation.
• Explorez Azure Monitor OpenTelemetry pour une surveillance complète des applications.