Behandeln von Fehlern, die vom Azure SDK für Python erzeugt werden

Das Erstellen zuverlässiger Cloudanwendungen erfordert mehr als nur die Implementierung von Features. Sie erfordert auch robuste Fehlerbehandlungsstrategien. Wenn Sie mit verteilten Systemen und Clouddiensten arbeiten, muss Ihre Anwendung darauf vorbereitet sein, Fehlerszenarien ordnungsgemäß zu behandeln.

Das Azure SDK für Python bietet ein umfassendes Fehlermodell, das Entwicklern beim Erstellen robuster Anwendungen hilft. Das Verständnis dieses Fehlermodells ist von entscheidender Bedeutung für:

  • Verbesserung der Anwendungssicherheit durch Antizipieren und Behandeln häufiger Fehlerszenarien.
  • Verbesserung der Benutzererfahrung durch aussagekräftige Fehlermeldungen und sanfte Degradierung.
  • Vereinfachen der Problembehandlung durch Erfassen und Protokollieren relevanter Diagnoseinformationen.

In diesem Artikel wird die Fehlerarchitektur von Azure SDK für Python erläutert und praktische Anleitungen für die Implementierung einer effektiven Fehlerbehandlung in Ihren Anwendungen bereitgestellt.

Wie das Azure SDK für Python Fehler modelliert

Das Azure SDK für Python verwendet ein hierarchisches Ausnahmemodell, das allgemeine und spezifische Fehlerbehandlungsfunktionen bereitstellt. Der Kern dieses Modells ist AzureError, die als Basis-Ausnahmeklasse für alle Fehler im Zusammenhang mit dem Azure SDK dient.

Ausnahmehierarchie

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

Wichtige Ausnahmetypen

Fehler Description
AzureError Die Basis-Ausnahmeklasse für alle Azure SDK-Fehler. Verwenden Sie diese Ausnahme als Catchall, wenn Sie einen Azure-bezogenen Fehler behandeln müssen.
ClientAuthenticationError Wird ausgelöst, wenn die Authentifizierung fehlschlägt. Häufige Ursachen sind ungültige Anmeldeinformationen, abgelaufene Token und falsch konfigurierte Authentifizierungseinstellungen.
ResourceNotFoundError Wird ausgelöst, wenn versucht wird, auf eine Ressource zuzugreifen, die nicht vorhanden ist. Diese Ausnahme entspricht in der Regel HTTP 404-Antworten.
ResourceExistsError Wird ausgelöst, wenn Sie versuchen, eine ressource zu erstellen, die bereits vorhanden ist. Diese Ausnahme verhindert versehentliche Überschreibungen.
ServiceRequestError Wird ausgelöst, wenn das SDK keine Anforderung an den Dienst senden kann. Häufige Ursachen sind Netzwerkkonnektivitätsprobleme, Auflösungsfehler des Domänennamensystems und ungültige Dienstendpunkte.
ServiceResponseError Wird ausgelöst, wenn der Dienst eine unerwartete Antwort zurückgibt, die das SDK nicht verarbeiten kann.
HttpResponseError Wird für HTTP-Fehlerantworten ausgelöst (4xx- und 5xx-Statuscodes). Diese Ausnahme bietet Zugriff auf die zugrunde liegenden HTTP-Antwortdetails.

Häufige Fehlerszenarien

Wenn Sie typische Fehlerszenarien verstehen, können Sie geeignete Behandlungsstrategien für jede Situation implementieren.

Authentifizierungs- und Autorisierungsfehler

Authentifizierungsfehler treten auf, wenn das SDK Ihre Identität nicht überprüfen kann:

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

Autorisierungsfehler (in der Regel HttpResponseError mit 403 Status) treten auf, wenn Sie keine Berechtigungen besitzen:

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

Ressourcenfehler

Behandeln Sie fehlende Ressourcen sorgfältig:

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"

Verhindern der Erstellung doppelter Ressourcen:

from azure.core.exceptions import ResourceExistsError

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

Serverfehler

Behandeln Sie serverseitige Fehler entsprechend:

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

Bewährte Methoden für die Fehlerbehandlung

  • Verwenden Sie spezielle Ausnahmebehandlung: Fangen Sie immer bestimmte Ausnahmen ab, bevor Sie auf allgemeine Ausnahmen zurückgreifen:

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

  • Implementieren Sie geeignete Wiederholungsstrategien: Einige Fehler rechtfertigen Wiederholungsversuche, während andere nicht.

    Wiederholen Sie den Vorgang nicht:

    • 401 Nicht autorisiert (Authentifizierungsfehler)
    • 403 Verboten (Autorisierungsfehler)
    • 400 Ungültige Anforderung (Clientfehler)
    • 404 Nicht gefunden (es sei denn, Sie erwarten, dass die Ressource angezeigt wird)

    Erwägen Sie es erneut zu versuchen:

    • 408 Zeitüberschreitung bei Anforderung
    • 429 Zu viele Anfragen (mit geeigneter Wartezeit)
    • 500: Interner Serverfehler
    • 502 Ungültiges Gateway
    • 503 Dienst nicht verfügbar
    • Timeout für 504 Gateways

  • Extrahieren aussagekräftiger Fehlerinformationen

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

Wiederholungsrichtlinien und Resilienz

Das Azure SDK enthält integrierte Wiederholungsmechanismen, die vorübergehende Fehler automatisch behandeln.

Standard-Wiederholungsverhalten

Die meisten Azure SDK-Clients enthalten standardmäßige Wiederholungsrichtlinien, die:

  • Wiederholen Sie Versuche bei Verbindungsfehlern und bestimmten HTTP-Statuscodes.
  • Verwenden Sie exponentielles Backoff mit Jitter.
  • Beschränken Sie die Anzahl der Wiederholungsversuche.

Anpassen von Wiederholungsrichtlinien

Wenn das Standardverhalten ihrem Anwendungsfall nicht entspricht, können Sie die Wiederholungsrichtlinie anpassen:

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
)

Vermeiden Sie den Umgang mit Netzwerk- und Timeoutfehlern durch benutzerdefinierte Schleifen

Versuchen Sie, integrierte Wiederholungen für Netzwerk- und Timeoutfehler zu verwenden, bevor Sie Ihre eigene benutzerdefinierte Logik implementieren.

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

Implementieren von Schaltkreistrennmustern

Berücksichtigen Sie bei kritischen Vorgängen die Implementierung von Schaltkreistrennmustern:

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

Grundlegendes zu Fehlermeldungen und Codes

Azure-Dienste geben strukturierte Fehlerantworten zurück, die wertvolle Debuginformationen bereitstellen.

  • Analysieren von Fehlerantworten

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

  • Erfassen von Diagnoseinformationen: Erfassen Sie immer wichtige Diagnoseinformationen für die Problembehandlung:

    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

  • Protokollierung und Diagnose: Aktivieren Sie die Protokollierung auf SDK-Ebene, um detaillierte Problembehandlungen zu ermöglichen:

    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)

    Weitere Informationen zur Protokollierung finden Sie unter Konfigurieren der Protokollierung in den Azure-Bibliotheken für Python.

  • Verwenden der Netzwerkablaufverfolgung: Aktivieren Sie für das umfassende Debuggen die Ablaufverfolgung auf Netzwerkebene:

    Von Bedeutung

    Die HTTP-Protokollierung kann vertrauliche Informationen wie Kontoschlüssel in Headern und anderen Anmeldeinformationen enthalten. Achten Sie darauf, diese Protokolle zu schützen, um die Sicherheit nicht zu gefährden.

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

Besondere Überlegungen für die asynchrone Programmierung

Wenn Sie asynchrone Clients verwenden, erfordert die Fehlerbehandlung besondere Aufmerksamkeit.

  • Grundlegende asynchrone Fehlerbehandlung

    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

  • Behandeln von Stornierungen

    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

  • Gleichzeitige Fehlerbehandlung

    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

Zusammenfassung der bewährten Methoden

Eine effektive Fehlerbehandlung in Azure SDK für Python-Anwendungen erfordert Folgendes:

  • Fehler antizipieren: Cloudanwendungen müssen partielle Fehler ordnungsgemäß erwarten und behandeln.
  • Verwenden Sie spezielle Ausnahmebehandlung: Erfassen Sie bestimmte Ausnahmen wie ResourceNotFoundError und ClientAuthenticationError bevor Sie auf die allgemeine AzureError Behandlung zurückfallen.
  • Implementieren Sie intelligente Wiederholungslogik: Verwenden Sie integrierte Wiederholungsrichtlinien, oder passen Sie sie basierend auf Ihren Anforderungen an. Denken Sie daran, dass nicht alle Fehler Wiederholungen auslösen sollten.
  • Erfassen von Diagnoseinformationen: Protokollieren Sie immer Anforderungs-IDs, Fehlercodes und Zeitstempel für eine effektive Problembehandlung.
  • Geben Sie aussagekräftiges Benutzerfeedback an: Verwandeln Sie technische Fehler in benutzerfreundliche Nachrichten, während Sie technische Details für den Support bewahren.
  • Testfehlerszenarien: Schließen Sie die Fehlerbehandlung in Ihre Testabdeckung ein, um sicherzustellen, dass sich Ihre Anwendung unter Fehlerbedingungen ordnungsgemäß verhält.

Verwandte Inhalte

  • Zuletzt aktualisiert um