Gestire gli errori generati da Azure SDK per Python

La creazione di applicazioni cloud affidabili richiede più di implementare solo le funzionalità. Richiede anche strategie affidabili per la gestione degli errori. Quando si lavora con sistemi distribuiti e servizi cloud, l'applicazione deve essere preparata per gestire correttamente gli scenari di errore.

Azure SDK per Python offre un modello di errore completo progettato per aiutare gli sviluppatori a creare applicazioni resilienti. Comprendere questo modello di errore è fondamentale per:

• Miglioramento dell'affidabilità dell'applicazione anticipando e gestendo scenari di errore comuni.
• Miglioramento dell'esperienza utente tramite messaggi di errore significativi e degradazione progressiva.
• Semplificare la risoluzione dei problemi acquisendo e registrando le informazioni di diagnostica pertinenti.

Questo articolo illustra l'architettura degli errori di Azure SDK per Python e fornisce indicazioni pratiche per implementare una gestione efficace degli errori nelle applicazioni.

Come l'Azure SDK per Python modella gli errori

Azure SDK per Python usa un modello di eccezione gerarchico che fornisce funzionalità di gestione degli errori generali e specifiche. Al centro di questo modello è AzureError, che funge da classe di eccezione di base per tutti gli errori correlati ad Azure SDK.

Gerarchia delle eccezioni

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

Tipi di eccezione chiave

Errore	Description
AzureError	Classe di eccezione di base per tutti gli errori di Azure SDK. Usare questa eccezione come catchall quando è necessario gestire qualsiasi errore correlato ad Azure.
ClientAuthenticationError	Generato quando l'autenticazione non riesce. Le cause comuni includono credenziali non valide, token scaduti e impostazioni di autenticazione configurate in modo errato.
ResourceNotFoundError	Generato quando si tenta di accedere a una risorsa che non esiste. Questa eccezione corrisponde in genere alle risposte HTTP 404.
ResourceExistsError	Generato quando si tenta di creare una risorsa già esistente. Questa eccezione consente di evitare sovrascrizioni accidentali.
ServiceRequestError	Generato quando l'SDK non può inviare una richiesta al servizio. Le cause comuni includono problemi di connettività di rete, errori di risoluzione del sistema dei nomi di dominio ed endpoint di servizio non validi.
ServiceResponseError	Generato quando il servizio restituisce una risposta imprevista che l'SDK non può elaborare.
HttpResponseError	Generato per le risposte di errore HTTP (codici di stato 4xx e 5xx). Questa eccezione fornisce l'accesso ai dettagli della risposta HTTP sottostante.

Scenari di errore comuni

Comprendere gli scenari di errore tipici consente di implementare strategie di gestione appropriate per ogni situazione.

Errori di autenticazione e autorizzazione

Gli errori di autenticazione si verificano quando l'SDK non riesce a verificare l'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

Gli errori di autorizzazione, solitamente con stato 403 in HttpResponseError, si verificano quando non si dispone delle autorizzazioni:

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

Errori delle risorse

Gestire correttamente le risorse mancanti:

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"

Impedire la creazione di risorse duplicate:

from azure.core.exceptions import ResourceExistsError

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

Errori del server

Gestire gli errori lato server in modo appropriato:

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

Procedure consigliate per la gestione degli errori

• Usare una gestione delle eccezioni specifica: Intercettare sempre eccezioni specifiche prima di tornare a quelle generali:

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

• Implementare strategie di ripetizione appropriate: Alcuni errori giustificano nuovi tentativi, mentre altri no.

  Non riprovare:

  • 401 Non autorizzato (errori di autenticazione)
  • 403 Accesso negato (errori di autorizzazione)
  • 400 Richiesta non valida (errori client)
  • 404 Non trovato (a meno che tu non ti aspetti che venga visualizzata la risorsa)

  Considerare di riprovare:

  • 408 Timeout richiesta
  • 429 Troppe richieste (con backoff appropriato)
  • 500 Errore interno del server
  • 502 Gateway non valido
  • 503 Servizio non disponibile
  • Timeout del gateway 504

• Estrarre informazioni significative sugli errori

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

Strategie di ripetizione dei tentativi e resilienza

Azure SDK include meccanismi di ripetizione dei tentativi predefiniti che gestiscono automaticamente gli errori temporanei.

Comportamento di ripetizione dei tentativi predefinito

La maggior parte dei client Azure SDK include criteri di ripetizione dei tentativi predefiniti che:

• Riprova in caso di errori di connessione e codici di stato HTTP specifici.
• Usa backoff esponenziale con jitter.
• Limitare il numero di tentativi.

Personalizzare i criteri di ripetizione dei tentativi

Se il comportamento predefinito non è adatto al caso d'uso, è possibile personalizzare i criteri di ripetizione dei tentativi:

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
)

Evitare di gestire errori di rete e timeout con cicli personalizzati

Cerca di usare i ricorsi integrati per gli errori di rete e timeout prima di implementare la logica personalizzata.

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

Implementare modelli di interruttore

Per le operazioni critiche, è consigliabile implementare modelli di interruttore:

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

Informazioni su messaggi e codici di errore

I servizi di Azure restituiscono risposte di errore strutturate che forniscono informazioni preziose sul debug.

• Analizzare le risposte di errore

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

• Acquisire informazioni di diagnostica: Acquisire sempre le informazioni di diagnostica chiave per la risoluzione dei problemi:

  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
  

• Registrazione e diagnostica: Abilitare la registrazione a livello di SDK per la risoluzione dei problemi dettagliata:

  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)
  

  Per altre informazioni sulla registrazione, vedere Configurare la registrazione nelle librerie di Azure per Python.

• Usare la traccia di rete: Per il debug avanzato, abilitare la traccia a livello di rete:

  Importante

  La registrazione HTTP può includere informazioni riservate, ad esempio le chiavi dell'account nelle intestazioni e altre credenziali. Assicurarsi di proteggere questi log per evitare di compromettere la sicurezza.

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

Considerazioni speciali per la programmazione asincrona

Quando si usano client asincroni, la gestione degli errori richiede particolare attenzione.

• Gestione degli errori asincroni di 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
  

• Gestire gli annullamenti

  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
  

• Gestione degli errori simultanei

  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
  

Riepilogo delle procedure consigliate

Una gestione efficace degli errori nelle applicazioni Azure SDK per Python richiede di:

• Prevedere gli errori: Le applicazioni cloud devono aspettarsi e gestire correttamente errori parziali.
• Usare una gestione delle eccezioni specifica: Intercettare eccezioni specifiche come ResourceNotFoundError e ClientAuthenticationError prima di tornare alla gestione generale AzureError .
• Implementare la logica di ripetizione dei tentativi intelligenti: Usare criteri di ripetizione dei tentativi predefiniti o personalizzarli in base alle esigenze. Tenere presente che non tutti gli errori devono attivare nuovi tentativi.
• Acquisire informazioni di diagnostica: Registra sempre gli ID delle richieste, i codici di errore e i timestamp per una risoluzione efficace dei problemi.
• Fornire feedback significativo per gli utenti: Trasformate gli errori tecnici in messaggi a misura d'utente, preservando i dettagli tecnici per il supporto.
• Scenari di errore di test: Includere la gestione degli errori nella copertura dei test per assicurarsi che l'applicazione si comporti correttamente in condizioni di errore.

Contenuti correlati

• Esaminare il riferimento del modulo delle eccezioni di Azure Core.
• Informazioni sulla risoluzione dei problemi di autenticazione e autorizzazione.
• Esplora Azure Monitor OpenTelemetry per un monitoraggio esaustivo delle applicazioni.