Manipular erros produzidos pelo SDK do Azure para Python

A criação de aplicativos de nuvem confiáveis requer mais do que apenas implementar recursos. Ele também exige estratégias robustas de tratamento de erros. Quando você trabalha com sistemas distribuídos e serviços de nuvem, seu aplicativo deve estar preparado para lidar com cenários de falha normalmente.

O SDK do Azure para Python fornece um modelo de erro abrangente projetado para ajudar os desenvolvedores a criar aplicativos resilientes. Entender esse modelo de erro é crucial para:

• Melhorando a confiabilidade do aplicativo antecipando e tratando cenários comuns de falha.
• Melhorando a experiência do usuário por meio de mensagens de erro significativas e degradação elegante.
• Simplificando a solução de problemas capturando e registrando informações de diagnóstico relevantes.

Este artigo explora a arquitetura de erros do SDK do Azure para Python e fornece diretrizes práticas para implementar o tratamento eficaz de erros em seus aplicativos.

Como o SDK do Azure para modelos do Python erros

O SDK do Azure para Python usa um modelo de exceção hierárquica que fornece recursos gerais e específicos de tratamento de erros. No centro desse modelo está AzureError, que serve como a classe de exceção base para todos os erros relacionados ao SDK do Azure.

Hierarquia de exceções

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

Tipos de exceção de chave

Cenários de erro comuns

Entender cenários de erro típicos ajuda você a implementar estratégias de tratamento apropriadas para cada situação.

Erros de autenticação e autorização

Falhas de autenticação ocorrem quando o SDK não pode verificar sua identidade:

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

Erros de autorização (normalmente HttpResponseError com 403 status) ocorrem quando você não tem permissões:

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

Erros de recurso

Manipule recursos ausentes normalmente:

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"

Impedir a criação de recursos duplicados:

from azure.core.exceptions import ResourceExistsError

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

Erros de servidor

Tratar falhas do lado do servidor adequadamente:

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

Práticas recomendadas para tratamento de erros

• Use tratamento de exceção específico: Sempre capture exceções específicas antes de voltar para as gerais:

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

• Implementar estratégias de repetição apropriadas: Alguns erros justificam tentativas de repetição, enquanto outros não.

  Não tente novamente:

  • 401 Não autorizado (falhas de autenticação)
  • 403 Proibido (falhas de autorização)
  • 400 Solicitação Incorreta (erros do cliente)
  • 404 Não Encontrado (a menos que você espere que o recurso apareça)

  Considere tentar novamente em:

  • Tempo limite da solicitação 408
  • 429 Solicitações demais (com retirada apropriada)
  • 500 Erro interno do servidor
  • Gateway Inválido 502
  • Serviço 503 indisponível
  • Tempo limite do gateway 504

• Extrair informações de erro significativas

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

Políticas de repetição e resiliência

O SDK do Azure inclui mecanismos de repetição internos que lidam com falhas transitórias automaticamente.

Comportamento de repetição padrão

A maioria dos clientes do SDK do Azure inclui políticas de repetição padrão que:

• Tente novamente em caso de erros de conexão ou de códigos de status HTTP específicos.
• Use a retirada exponencial com tremulação.
• Limite o número de tentativas de repetição.

Personalizar políticas de repetição

Se o comportamento padrão não atender ao seu caso de uso, você poderá personalizar a política de repetição:

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
)

Evite lidar com erros de rede e tempo limite com loops personalizados

Tente usar retries internos para erros de rede e tempo limite antes de implementar sua própria lógica personalizada.

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

Implementar padrões de disjuntor

Para operações críticas, considere implementar padrões de disjuntor:

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

Entender mensagens de erro e códigos

Os serviços do Azure retornam respostas de erro estruturadas que fornecem informações valiosas de depuração.

• Analisar respostas de erro

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

• Capturar informações de diagnóstico: Sempre capture as principais informações de diagnóstico para solução de problemas:

  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
  

• Log e diagnóstico: Habilite o registro em log no nível do SDK para solução de problemas detalhada:

  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)
  

  Para obter mais informações sobre o registro em log, consulte Configurar o log nas bibliotecas do Azure para Python.

• Use o rastreamento de rede: Para depuração profunda, habilite o rastreamento em nível de rede:

  Importante

  O log HTTP pode incluir informações confidenciais, como chaves de conta em cabeçalhos e outras credenciais. Certifique-se de proteger esses logs para evitar comprometer a segurança.

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

Considerações especiais para programação assíncrona

Quando você usa clientes assíncronos, o tratamento de erros requer atenção especial.

• Tratamento de erros assíncronos básicos

  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
  

• Manipular cancelamentos

  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
  

• Tratamento de erros simultâneos

  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
  

Resumo das práticas recomendadas

O tratamento eficaz de erros no SDK do Azure para aplicativos Python exige que você:

• Prever falhas: Os aplicativos de nuvem devem esperar e lidar com falhas parciais normalmente.
• Use tratamento de exceção específico: Capture exceções específicas como ResourceNotFoundError e ClientAuthenticationError antes de recorrer ao tratamento geral AzureError.
• Implementar lógica de repetição inteligente: Use políticas de repetição internas ou personalize-as com base em suas necessidades. Lembre-se de que nem todos os erros devem disparar novas tentativas.
• Capturar informações de diagnóstico: Sempre registre IDs de solicitação, códigos de erro e carimbos de data/hora para solução de problemas efetiva.
• Forneça comentários significativos do usuário: Transforme erros técnicos em mensagens amigáveis ao preservar detalhes técnicos para suporte.
• Cenários de erro de teste: Inclua o tratamento de erros na cobertura de teste para garantir que seu aplicativo se comporte corretamente em condições de falha.

Conteúdo relacionado

• Examine a referência do módulo de exceções do Azure Core.
• Saiba mais sobre como solucionar problemas de autenticação e autorização.
• Explore o OpenTelemetry do Azure Monitor para monitoramento abrangente de aplicativos.