Lidar com erros produzidos pelo Azure SDK para Python

Construir aplicações cloud fiáveis requer mais do que apenas implementar funcionalidades. Também exige estratégias robustas de gestão de erros. Quando trabalha com sistemas distribuídos e serviços na cloud, a sua aplicação deve estar preparada para lidar com cenários de falha com elegância.

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

Melhorar a fiabilidade da aplicação antecipando e lidando com cenários comuns de falha.
Melhorar a experiência do utilizador através de mensagens de erro significativas e degradação gradual.
Simplificar a resolução de problemas através da captura e registo de informações de diagnóstico relevantes.

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

Como o SDK do Azure para Python modela erros

O Azure SDK para Python utiliza um modelo hierárquico de exceções que fornece capacidades gerais e específicas de tratamento de erros. No núcleo deste modelo está AzureError, que serve como a classe base de exceção para todos os erros relacionados com o Azure SDK.

Hierarquia de exceção

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

Principais tipos de exceção

Erro	Description
`AzureError`	A classe de exceção base para todos os erros do SDK do Azure. Use esta exceção como um termo genérico quando precisar de lidar com qualquer erro relacionado com o Azure.
`ClientAuthenticationError`	Gerado quando a autenticação falha. As causas comuns incluem credenciais inválidas, tokens expirados e configurações de autenticação mal configuradas.
`ResourceNotFoundError`	Gerado ao tentar acessar um recurso que não existe. Esta exceção corresponde tipicamente a respostas HTTP 404.
`ResourceExistsError`	Gerado ao tentar criar um recurso que já existe. Esta exceção ajuda a evitar sobrescrições acidentais.
`ServiceRequestError`	Gerado quando o SDK não pode enviar uma solicitação para o serviço. As causas comuns incluem problemas de conectividade de rede, falhas na resolução do Sistema de Nomes de Domínio e endpoints de serviço inválidos.
`ServiceResponseError`	Gerado quando o serviço retorna uma resposta inesperada que o SDK não pode processar.
`HttpResponseError`	Gerado para respostas de erro HTTP (códigos de status 4xx e 5xx). Esta exceção fornece acesso aos detalhes da resposta HTTP subjacente.

Cenários de erro comuns

Compreender os cenários de erro típicos ajuda a implementar estratégias de manipulação 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 (tipicamente HttpResponseError com o estado 403) ocorrem quando 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 recursos

Lide com 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"

Evite 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 do servidor

Lidar com 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 específico de exceções: Apanhe sempre exceções específicas antes de recorrer às mais 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}")

Implemente estratégias adequadas de repetição: 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 Bad Request (erros do cliente)
- 404 Não encontrado (a menos que você espere que o recurso apareça)
Considere tentar novamente:
- 408 Tempo limite de solicitação
- 429 Demasiados Pedidos (com recuo adequado)
- 500 Erro interno do servidor
- 502 Gateway ruim
- 503 Serviço indisponível
- 504 Tempo limite do gateway

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 internos de repetição 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:

Tentar novamente em caso de erros de ligação e códigos de estado HTTP específicos.
Usa um retrocesso exponencial com jitter.
Limite o número de tentativas de repetição.

Personalizar políticas de repetição

Se o comportamento padrão não for adequado ao seu caso de uso, pode personalizar a política de tentativa de nova:

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 reintentos incorporados para erros de rede e timeout antes de implementar a 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 disjuntores

Para operações críticas, considere a implementação de padrões de disjuntores:

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

Compreender mensagens de erro e códigos

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

Respostas de erro de parsing

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

Recolha informação de diagnóstico: Recolha sempre informações de diagnóstico essenciais para resoluçã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

Registo e diagnóstico: Ative o registo ao nível do SDK para resolução detalhada de problemas:

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 log, consulte Configurar o log nas bibliotecas do Azure para Python.

Utilize o rastreamento de rede: Para depuração avançada, ative o rastreamento a 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 utiliza clientes assíncronos, o tratamento de erros requer atenção especial.

Tratamento de erros assíncrono básico

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

Gerir 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 simultâneo de erros

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 melhores práticas

O tratamento eficaz de erros no Azure SDK para aplicações Python requer que:

Antecipe falhas: As aplicações na cloud devem esperar e lidar com falhas parciais de forma elegante.
Use tratamento específico de exceções: Intercepte exceções específicas, como ResourceNotFoundError e ClientAuthenticationError antes de retornar ao tratamento geral AzureError.
Implemente uma lógica inteligente de repetição: Utilize políticas de repetição integradas ou personalize-as de acordo com as suas necessidades. Lembre-se de que nem todos os erros devem desencadear tentativas.
Recolha informação de diagnóstico: Registe sempre os IDs dos pedidos, códigos de erro e carimbos de data para uma resolução eficaz de problemas.
Forneça feedback significativo dos utilizadores: Transforme erros técnicos em mensagens intuitivas enquanto preserva detalhes técnicos para suporte.
Cenários de erro de teste: Inclua o tratamento de erros na sua cobertura de testes para garantir que a sua aplicação se comporta corretamente em condições de falha.

Consulte a referência do módulo de exceções Azure Core.
Aprenda sobre a resolução de problemas de autenticação e autorização.
Explore o Azure Monitor OpenTelemetry para uma monitorização abrangente de aplicações.

Comentários

Esta página foi útil?

Última atualização a 2025-12-09