Compartir a través de


Control de errores generados por el SDK de Azure para Python

La creación de aplicaciones en la nube confiables requiere algo más que implementar características. También exige estrategias sólidas de control de errores. Al trabajar con sistemas distribuidos y servicios en la nube, la aplicación debe estar preparada para controlar los escenarios de error correctamente.

El SDK de Azure para Python proporciona un modelo de error completo diseñado para ayudar a los desarrolladores a crear aplicaciones resistentes. Comprender este modelo de error es fundamental para:

  • Mejorar la confiabilidad de las aplicaciones anticipando y controlando escenarios de error comunes.
  • Mejora de la experiencia del usuario a través de mensajes de error significativos y degradación suave.
  • Simplifique la solución de problemas mediante la captura y el registro de información de diagnóstico pertinente.

En este artículo se explora la arquitectura de errores del SDK de Azure para Python y se proporcionan instrucciones prácticas para implementar un control eficaz de errores en las aplicaciones.

Cómo se producen errores del SDK de Azure para Python

El SDK de Azure para Python usa un modelo de excepción jerárquico que proporciona funcionalidades generales y específicas de control de errores. En el núcleo de este modelo es AzureError, que actúa como clase de excepción base para todos los errores relacionados con el SDK de Azure.

Jerarquía de excepciones

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

Tipos de excepciones clave

Error Description
AzureError Clase de excepción base para todos los errores del SDK de Azure. Utilice esta excepción como un "catch-all" cuando necesite manejar cualquier error relacionado con Azure.
ClientAuthenticationError Se genera cuando se produce un error en la autenticación. Entre las causas comunes se incluyen credenciales no válidas, tokens expirados y configuraciones de autenticación mal configuradas.
ResourceNotFoundError Se genera al intentar acceder a un recurso que no existe. Esta excepción normalmente corresponde a respuestas HTTP 404.
ResourceExistsError Se genera al intentar crear un recurso que ya existe. Esta excepción ayuda a evitar sobrescrituras accidentales.
ServiceRequestError Se genera cuando el SDK no puede enviar una solicitud al servicio. Entre las causas comunes se incluyen problemas de conectividad de red, errores de resolución del sistema de nombres de dominio y puntos de conexión de servicio no válidos.
ServiceResponseError Se genera cuando el servicio devuelve una respuesta inesperada que el SDK no puede procesar.
HttpResponseError Se genera para las respuestas de error HTTP (códigos de estado 4xx y 5xx). Esta excepción proporciona acceso a los detalles de respuesta HTTP subyacentes.

Escenarios de error comunes

Comprender los escenarios de error típicos le ayuda a implementar estrategias de control adecuadas para cada situación.

Errores de autenticación y autorización

Los errores de autenticación se producen cuando el SDK no puede comprobar la identidad:

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

Los errores de autorización (normalmente HttpResponseError con 403 estado) se producen cuando faltan permisos:

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

Errores de recursos

Controle los recursos que faltan correctamente:

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 la creación de recursos duplicados:

from azure.core.exceptions import ResourceExistsError

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

Errores de servidor

Controle correctamente los errores del lado servidor:

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

Procedimientos recomendados para el control de errores

  • Use un control de excepciones específico: Detectar siempre excepciones específicas antes de volver a las generales:

    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}")
    
  • Implemente las estrategias de reintento adecuadas: Algunos errores garantizan los reintentos, mientras que otros no.

    No vuelva a intentarlo:

    • 401 No autorizado (errores de autenticación)
    • 403 Prohibido (errores de autorización)
    • 400 Solicitud incorrecta (errores de cliente)
    • 404 No encontrado (a menos que espere que aparezca el recurso)

    Considere la posibilidad de volver a intentarlo:

    • 408 Tiempo de espera de solicitud
    • 429 Demasiadas solicitudes (con el retroceso adecuado)
    • Error interno del servidor 500
    • 502 Puerta de enlace incorrecta
    • 503 Servicio no disponible
    • Tiempo de espera de puerta de enlace 504
  • Extracción de información de error significativa

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

Directivas de reintento y resistencia

El SDK de Azure incluye mecanismos de reintento integrados que controlan los errores transitorios automáticamente.

Comportamiento de reintento predeterminado

La mayoría de los clientes del SDK de Azure incluyen directivas de reintento predeterminadas que:

  • Vuelva a intentar errores de conexión y códigos de estado HTTP específicos.
  • Use retroceso exponencial con vibración.
  • Limite el número de reintentos.

Personalización de directivas de reintento

Si el comportamiento predeterminado no se ajusta a su caso de uso, puede personalizar la directiva de reintento:

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
)

Evitar el control de errores de red y tiempo de espera con bucles personalizados

Intente usar reintentos integrados para errores de red y tiempo de espera antes de implementar su propia 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

Implemente patrones de cortacircuitos

Para las operaciones críticas, considere la posibilidad de implementar patrones de disyuntor:

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

Descripción de los mensajes de error y los códigos

Los servicios de Azure devuelven respuestas de error estructuradas que proporcionan información de depuración valiosa.

  • Análisis de respuestas de error

    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 información de diagnóstico: Capture siempre la información de diagnóstico clave para solucionar 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
    
  • Registro y diagnóstico: Habilite el registro de nivel de SDK para solucionar problemas detallados:

    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 más información sobre el registro, consulte Configuración del registro en las bibliotecas de Azure para Python.

  • Use el seguimiento de red: Para la depuración profunda, habilite el seguimiento de nivel de red:

    Importante

    El registro HTTP puede incluir información confidencial, como claves de cuenta en encabezados y otras credenciales. Asegúrese de proteger estos registros para evitar poner en peligro la seguridad.

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

Consideraciones especiales para la programación asincrónica

Cuando se usan clientes asincrónicos, el control de errores requiere especial atención.

  • Control básico de errores asincrónicos

    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
    
  • Control de cancelaciones

    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
    
  • Control simultáneo de errores

    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
    

Resumen de los procedimientos recomendados

El control de errores efectivo en azure SDK para aplicaciones de Python requiere que:

  • Anticipación de errores: Las aplicaciones en la nube deben esperar y controlar los errores parciales correctamente.
  • Use un control de excepciones específico: Capture excepciones específicas como ResourceNotFoundError y ClientAuthenticationError antes de volver al control general AzureError .
  • Implemente la lógica de reintento inteligente: Use directivas de reintento integradas o personalícelas en función de sus necesidades. Recuerde que no todos los errores deben desencadenar reintentos.
  • Capturar información de diagnóstico: Registre siempre los identificadores de solicitud, los códigos de error y las marcas de tiempo para solucionar problemas eficaces.
  • Proporcione comentarios significativos del usuario: Transforme los errores técnicos en mensajes fáciles de usar mientras conserva los detalles técnicos para obtener soporte técnico.
  • Escenarios de error de prueba: Incluya el control de errores en la cobertura de pruebas para asegurarse de que la aplicación se comporta correctamente en condiciones de error.