Obsługa błędów generowanych przez zestaw Azure SDK dla języka Python

Tworzenie niezawodnych aplikacji w chmurze wymaga więcej niż tylko implementowania funkcji. Wymaga również niezawodnych strategii obsługi błędów. Podczas pracy z systemami rozproszonymi i usługami w chmurze aplikacja musi być przygotowana do bezproblemowego obsługi scenariuszy awarii.

Zestaw Azure SDK dla języka Python udostępnia kompleksowy model błędów, który ułatwia deweloperom tworzenie odpornych aplikacji. Zrozumienie tego modelu błędów ma kluczowe znaczenie dla:

Zwiększanie niezawodności aplikacji dzięki przewidywaniu i obsłudze typowych scenariuszy awarii.
Ulepszanie doświadczenia użytkownika dzięki znaczącym komunikatom o błędach i stopniowemu pogorszeniu funkcjonalności.
Upraszczanie rozwiązywania problemów przez przechwytywanie i rejestrowanie odpowiednich informacji diagnostycznych.

W tym artykule omówiono architekturę błędów zestawu Azure SDK dla języka Python i przedstawiono praktyczne wskazówki dotyczące implementowania efektywnej obsługi błędów w aplikacjach.

Jak występują błędy zestawu Azure SDK dla modeli języka Python

Zestaw Azure SDK dla języka Python używa hierarchicznego modelu wyjątków, który zapewnia ogólne i specyficzne możliwości obsługi błędów. Podstawowym elementem tego modelu jest AzureError, który służy jako klasa wyjątku podstawowego dla wszystkich błędów związanych z zestawem Azure SDK.

Hierarchia wyjątków

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

Typy wyjątków klucza

Error	Description
`AzureError`	Klasa wyjątku podstawowego dla wszystkich błędów zestawu Azure SDK. Użyj tego wyjątku jako uniwersalnego, jeśli musisz obsłużyć dowolny błąd związany z platformą Azure.
`ClientAuthenticationError`	Zgłaszane, gdy uwierzytelnianie kończy się niepowodzeniem. Typowe przyczyny obejmują nieprawidłowe poświadczenia, wygasłe tokeny i błędnie skonfigurowane ustawienia uwierzytelniania.
`ResourceNotFoundError`	Zgłoszone podczas próby uzyskania dostępu do zasobu, który nie istnieje. Ten wyjątek zazwyczaj odpowiada odpowiedziom HTTP 404.
`ResourceExistsError`	Zgłoszone podczas próby utworzenia zasobu, który już istnieje. Ten wyjątek pomaga zapobiec przypadkowemu nadpisywaniu.
`ServiceRequestError`	Zgłaszane, gdy zestaw SDK nie może wysłać żądania do usługi. Typowe przyczyny obejmują problemy z łącznością sieciową, błędy rozpoznawania nazw domen i nieprawidłowe punkty końcowe usługi.
`ServiceResponseError`	Zgłaszane, gdy usługa zwraca nieoczekiwaną odpowiedź, że zestaw SDK nie może przetworzyć.
`HttpResponseError`	Zgłoszone w przypadku odpowiedzi na błędy HTTP (kody stanu 4xx i 5xx). Ten wyjątek zapewnia dostęp do podstawowych szczegółów odpowiedzi HTTP.

Typowe scenariusze błędów

Zrozumienie typowych scenariuszy błędów pomaga zaimplementować odpowiednie strategie obsługi dla każdej sytuacji.

Błędy uwierzytelniania i autoryzacji

Błędy uwierzytelniania występują, gdy zestaw SDK nie może zweryfikować twojej tożsamości:

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

Błędy autoryzacji (zazwyczaj ze statusem HttpResponseError) występują, gdy brak Ci uprawnień 403.

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

Błędy zasobów

Obsługa brakujących zasobów w sposób bezproblemowy:

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"

Zapobiegaj tworzeniu zduplikowanych zasobów:

from azure.core.exceptions import ResourceExistsError

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

Błędy serwera

Odpowiednio obsłuż błędy po stronie serwera:

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

Najlepsze rozwiązania dotyczące obsługi błędów

Użyj określonej obsługi wyjątków: Zawsze przechwytuj określone wyjątki przed powrotem do ogólnych:

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

Zaimplementuj odpowiednie strategie ponawiania prób: Niektóre błędy uzasadniają ponawianie prób, podczas gdy inne nie.

Nie próbuj ponownie:
- 401 Brak autoryzacji (błędy uwierzytelniania)
- 403 Zabronione (błędy autoryzacji)
- 400 Nieprawidłowe żądanie (błędy klienta)
- 404 Nie znaleziono (chyba że zostanie wyświetlony zasób)
Rozważ ponowienie próby:
- 408 Limit czasu żądania
- 429 Zbyt wiele żądań (z odpowiednim wycofywaniem)
- 500 Wewnętrzny błąd serwera
- 502 — zła brama
- Usługa 503 jest niedostępna
- 504 Limit czasu bramy

Wyodrębnianie znaczących informacji o błędzie

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

Zasady ponawiania prób i odporność

Zestaw Azure SDK zawiera wbudowane mechanizmy ponawiania, które automatycznie obsługują błędy przejściowe.

Domyślne zachowanie ponawiania prób

Większość klientów zestawu Azure SDK obejmuje domyślne zasady ponawiania prób, które:

Ponów próbę w przypadku błędów połączenia i określonych kodów stanu HTTP.
Użyj wycofywania wykładniczego z roztrzaskiem.
Ogranicz liczbę ponownych prób.

Dostosowywanie zasad ponawiania prób

Jeśli domyślne zachowanie nie pasuje do twojego przypadku użycia, możesz dostosować zasady ponawiania prób:

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
)

Unikaj obsługi błędów sieci i limitu czasu w pętlach niestandardowych

Spróbuj użyć wbudowanych ponownych prób dla błędów sieci i przekroczenia limitu czasu przed zaimplementowaniem własnej logiki niestandardowej.

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

Implementowanie wzorców wyłącznika

W przypadku operacji krytycznych rozważ zaimplementowanie wzorców wyłącznika:

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

Omówienie komunikatów o błędach i kodów

Usługi platformy Azure zwracają ustrukturyzowane odpowiedzi o błędach, które zapewniają cenne informacje dotyczące debugowania.

Analizowanie odpowiedzi na błędy

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

Przechwyć informacje diagnostyczne: Zawsze przechwytuj kluczowe informacje diagnostyczne na potrzeby rozwiązywania problemów:

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

Rejestrowanie i diagnostyka: Włącz rejestrowanie na poziomie zestawu SDK, aby uzyskać szczegółowe informacje na temat rozwiązywania problemów:

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)

Aby uzyskać więcej informacji na temat rejestrowania, zobacz Konfigurowanie rejestrowania w bibliotekach platformy Azure dla języka Python.

Użyj śledzenia sieci: W przypadku głębokiego debugowania włącz śledzenie na poziomie sieci:

Ważne

Rejestrowanie HTTP może zawierać poufne informacje, takie jak klucze konta w nagłówkach i inne poświadczenia. Pamiętaj, aby chronić te dzienniki, aby uniknąć naruszenia zabezpieczeń.

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

Specjalne zagadnienia dotyczące programowania asynchronicznego

W przypadku korzystania z klientów asynchronicznych obsługa błędów wymaga szczególnej uwagi.

Podstawowa obsługa błędów asynchronicznych

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

Obsługa anulowania

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

Współbieżna obsługa błędów

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

Podsumowanie najlepszych rozwiązań

Efektywna obsługa błędów w Azure SDK dla aplikacji Python wymaga:

Przewidywanie błędów: Aplikacje w chmurze muszą bezpiecznie oczekiwać i obsługiwać częściowe błędy.
Użyj określonej obsługi wyjątków: Przechwyć określone wyjątki, takie jak ResourceNotFoundError i ClientAuthenticationError przed powrotem do ogólnej AzureError obsługi.
Zaimplementuj inteligentną logikę ponawiania: Użyj wbudowanych zasad ponawiania lub dostosuj je w zależności od potrzeb. Pamiętaj, że nie wszystkie błędy powinny wyzwalać ponawianie prób.
Przechwyć informacje diagnostyczne: Zawsze rejestruj identyfikatory żądań, kody błędów i sygnatury czasowe, aby skutecznie rozwiązywać problemy.
Podaj znaczącą informację zwrotną dla użytkownika: Przekształcaj błędy techniczne w przyjazne komunikaty dla użytkownika, zachowując szczegóły techniczne dla wsparcia.
Scenariusze błędów testowych: Uwzględnij obsługę błędów w zakresie testów, aby upewnić się, że aplikacja działa prawidłowo w warunkach awarii.

Zapoznaj się z dokumentacją modułu wyjątków platformy Azure Core.
Dowiedz się więcej o rozwiązywaniu problemów z uwierzytelnianiem i autoryzacją.
Zapoznaj się z usługą Azure Monitor OpenTelemetry , aby uzyskać kompleksowe monitorowanie aplikacji.

Opinia

Czy ta strona była pomocna?

Last updated on 2025-12-09