Vyřešte chyby produkované sadou Azure SDK pro Python

Vytváření spolehlivých cloudových aplikací vyžaduje více než jenom implementaci funkcí. Vyžaduje také robustní strategie zpracování chyb. Při práci s distribuovanými systémy a cloudovými službami musí být vaše aplikace připravená na řádné zpracování scénářů selhání.

Sada Azure SDK pro Python poskytuje komplexní model chyb navržený tak, aby vývojářům pomohl vytvářet odolné aplikace. Pochopení tohoto modelu chyb je zásadní pro:

• Zlepšení spolehlivosti aplikací předvídáním a zpracováním běžných scénářů selhání
• Vylepšení uživatelského prostředí prostřednictvím smysluplných chybových zpráv a postupného snižování funkcionalit.
• Zjednodušení řešení potíží zachytáváním a protokolováním relevantních diagnostických informací

Tento článek popisuje architekturu chyb v sadě Azure SDK pro Python a poskytuje praktické pokyny k implementaci efektivního zpracování chyb ve vašich aplikacích.

Chyby modelů Sady Azure SDK pro Python

Sada Azure SDK pro Python používá hierarchický model výjimek, který poskytuje obecné a specifické možnosti zpracování chyb. Jádrem tohoto modelu je AzureError, který slouží jako základní třída výjimek pro všechny chyby související se sadou Azure SDK.

Hierarchie výjimek

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

Klíčové typy výjimek

Běžné chybové scénáře

Pochopení typických chybových scénářů vám pomůže implementovat vhodné strategie zpracování pro každou situaci.

Chyby ověřování a autorizace

K selhání ověřování dochází v případě, že sada SDK nemůže ověřit vaši identitu:

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

K chybám autorizace (obvykle HttpResponseError se stavem 403 ) dochází v případě, že nemáte oprávnění:

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

Chyby prostředků

Zpracování chybějících prostředků bez odkladu:

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"

Zabránit vytvoření duplicitního prostředku:

from azure.core.exceptions import ResourceExistsError

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

Chyby serveru

Odpovídající zpracování selhání na straně serveru:

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

Osvědčené postupy pro zpracování chyb

• Použijte konkrétní zpracování výjimek: Před návratem k obecným výjimkám vždy zachyťte konkrétní výjimky:

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

• Implementace vhodných strategií opakování: Některé chyby vyžadují pokusy o opakování, zatímco jiné ne.

  Zkuste to znovu:

  • 401 Neautorizováno (selhání ověřování)
  • 403 Zakázáno (chyby autorizace)
  • 400 Chybný požadavek (chyby klienta)
  • 404 Nenalezena (pokud neočekáváte, že se prostředek zobrazí)

  Zvažte opakování:

  • 408 Vypršení časového limitu požadavku
  • 429 Příliš mnoho požadavků (s odpovídajícími omezeními)
  • 500 – Vnitřní chyba serveru
  • 502 – Chybná brána
  • 503 Služba není k dispozici
  • 504 Vypršení časového limitu brány

• Extrahování smysluplných informací o chybách

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

Zásady opakování a odolnost

Sada Azure SDK obsahuje integrované mechanismy opakování, které automaticky zpracovávají přechodné selhání.

Výchozí chování opakování

Většina klientů Sady Azure SDK obsahuje výchozí zásady opakování, které:

• Znovu se pokuste při chybách připojení a konkrétních stavových kódech HTTP.
• Používejte exponenciální ústup s jitterem.
• Omezte počet opakovaných pokusů.

Přizpůsobení zásad opakování

Pokud výchozí chování nevyhovuje vašemu případu použití, můžete upravit zásadu opakování:

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
)

Vyhněte se zpracování chyb sítě a vypršení časového limitu s využitím vlastních smyček

Před implementací vlastní logiky se pokuste použít předdefinované opakování pro chyby sítě a časového limitu.

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

Implementace vzorů přerušovače obvodů

U kritických operací zvažte implementaci vzorů jističe:

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

Vysvětlení chybových zpráv a kódů

Služby Azure vrací strukturované chybové odpovědi, které poskytují cenné informace o ladění.

• Analýza chybových odpovědí

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

• Zachytávání diagnostických informací: Vždy zachytávat klíčové diagnostické informace pro řešení potíží:

  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
  

• Protokolování a diagnostika: Zapněte protokolování na úrovni sady SDK pro podrobné řešení potíží.

  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)
  

  Další informace o protokolování najdete v tématu Konfigurace protokolování v knihovnách Azure pro Python.

• Použijte trasování sítě: Pro podrobné ladění povolte trasování sítě na úrovni protokolu:

  Důležité

  Protokolování HTTP může obsahovat citlivé informace, jako jsou klíče účtu, do hlaviček a dalších přihlašovacích údajů. Nezapomeňte tyto protokoly chránit, abyste se vyhnuli ohrožení zabezpečení.

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

Zvláštní aspekty asynchronního programování

Při použití asynchronních klientů vyžaduje zpracování chyb zvláštní pozornost.

• Základní zpracování asynchronních chyb

  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
  

• Zpracování zrušení

  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
  

• Souběžné zpracování chyb

  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
  

Shrnutí osvědčených postupů

Efektivní zpracování chyb v aplikacích Azure SDK pro Python vyžaduje, abyste:

• Předvídání selhání: Cloudové aplikace musí očekávat a zpracovávat částečná selhání elegantně.
• Použijte konkrétní zpracování výjimek: Zachyťte konkrétní výjimky jako ResourceNotFoundError a ClientAuthenticationError před návratem do obecného AzureError zpracování.
• Implementace logiky inteligentního opakování: Použijte předdefinované zásady opakování nebo je upravte podle svých potřeb. Mějte na paměti, že ne všechny chyby by měly aktivovat opakování.
• Zachytávání diagnostických informací: Vždy protokolujte ID žádostí, kódy chyb a časová razítka pro efektivní řešení potíží.
• Poskytněte smysluplnou zpětnou vazbu uživatelů: Při zachování technických podrobností pro podporu můžete převést technické chyby na uživatelsky přívětivé zprávy.
• Testovací chybové scénáře: Zahrňte do pokrytí testu zpracování chyb, abyste měli jistotu, že se vaše aplikace chová správně za podmínek selhání.

Související obsah

• Projděte si referenční informace k modulu výjimek Azure Core.
• Přečtěte si o řešení potíží s ověřováním a autorizací.
• Prozkoumejte Azure Monitor OpenTelemetry pro komplexní monitorování aplikací.