Python için Azure SDK tarafından oluşturulan hataları işleme

Güvenilir bulut uygulamaları oluşturmak için yalnızca özellikleri uygulamaktan fazlası gerekir. Ayrıca güçlü hata işleme stratejileri de talep eder. Dağıtılmış sistemler ve bulut hizmetleriyle çalışırken uygulamanızın hata senaryolarını düzgün bir şekilde işlemeye hazır olması gerekir.

Python için Azure SDK, geliştiricilerin dayanıklı uygulamalar oluşturmasına yardımcı olmak için tasarlanmış kapsamlı bir hata modeli sağlar. Bu hata modelini anlamak aşağıdakiler için çok önemlidir:

• Yaygın hata senaryolarını tahmin ederek ve işleyerek uygulama güvenilirliğini artırma.
• Anlamlı hata iletileri ve zarif bir şekilde azalma yoluyla kullanıcı deneyimini geliştirme.
• İlgili tanılama bilgilerini yakalayıp günlüğe kaydetme yoluyla sorun gidermeyi basitleştirme.

Bu makalede Python'ın hata mimarisi için Azure SDK'sı incelenmektedir ve uygulamalarınızda etkili hata işlemeyi uygulamaya yönelik pratik yönergeler sağlanmaktadır.

Python için Azure SDK'sı hataları nasıl modeller?

Python için Azure SDK, genel ve belirli hata işleme özellikleri sağlayan hiyerarşik bir özel durum modeli kullanır. Bu modelin çekirdeğinde, AzureErrorAzure SDK ile ilgili tüm hatalar için temel özel durum sınıfı görevi görür.

Özel durum hiyerarşisi

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

Önemli özel durum türleri

Yaygın hata senaryoları

Tipik hata senaryolarını anlamak, her durum için uygun işleme stratejileri uygulamanıza yardımcı olur.

Kimlik doğrulama ve yetkilendirme hataları

KIMLIK doğrulama hataları, SDK kimliğinizi doğrulayamazsa oluşur:

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

Yetkilendirme hataları (genellikle HttpResponseError durumuyla birlikte 403) izinleriniz olmadığında oluşur.

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

Kaynak hataları

Eksik kaynakları düzgün bir şekilde işleyebilir:

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"

Yinelenen kaynak oluşturmayı engelle:

from azure.core.exceptions import ResourceExistsError

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

Sunucu hataları

Sunucu tarafı hatalarını uygun şekilde işleyin:

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

Hata işleme için en iyi yöntemler

• Özel durum işlemeyi kullanın: Genel özel durumlara geri dönmeden önce her zaman belirli özel durumları yakalayın:

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

• Uygun yeniden deneme stratejilerini uygulayın: Bazı hatalar yeniden deneme girişimlerine neden olurken, diğerleri denemez.

  Şu şekilde yeniden deneme:

  • 401 Yetkisiz (kimlik doğrulama hataları)
  • 403 Yasak (yetkilendirme hataları)
  • 400 Hatalı İstek (istemci hataları)
  • 404 Bulunamadı (kaynağın görünmesini beklemediğiniz sürece)

  Yeniden denemeyi göz önünde bulundurun:

  • 408 İstek Zaman Aşımı
  • 429 Çok Fazla İstek (uygun geri alma ile)
  • 500 İç Sunucu Hatası
  • 502 Hatalı Ağ Geçidi
  • 503 Hizmet Kullanılamıyor
  • 504 Ağ Geçidi Zaman Aşımı

• Anlamlı hata bilgilerini ayıklama

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

İlkeleri ve dayanıklılığı yeniden deneme

Azure SDK, geçici hataları otomatik olarak işleyen yerleşik yeniden deneme mekanizmaları içerir.

Varsayılan yeniden deneme davranışı

Azure SDK istemcilerinin çoğu aşağıdaki varsayılan yeniden deneme ilkelerini içerir:

• Bağlantı hatalarını ve belirli HTTP durum kodlarını yeniden deneyin.
• Üstel geri çekilme ile rastgelelik kullanın.
• Yeniden deneme denemesi sayısını sınırlayın.

Yeniden deneme ilkelerini özelleştirme

Varsayılan davranış kullanım örneğiniz için uygun değilse yeniden deneme ilkesini özelleştirebilirsiniz:

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
)

Özel döngülerle ağ ve zaman aşımı hatalarını işlemekten kaçının

Kendi özel mantığınızı uygulamadan önce ağ ve zaman aşımı hataları için yerleşik yeniden denemeleri kullanmayı deneyin.

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

Devre kesici desenlerini uygulama

Kritik işlemler için devre kesici desenleri uygulamayı göz önünde bulundurun:

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

Hata iletilerini ve kodları anlama

Azure hizmetleri değerli hata ayıklama bilgileri sağlayan yapılandırılmış hata yanıtları döndürür.

• Hata yanıtlarını ayrıştırma

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

• Tanılama bilgilerini yakalama: Sorun giderme için her zaman önemli tanılama bilgilerini yakalayın:

  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
  

• Günlüğe kaydetme ve tanılama: Ayrıntılı sorun giderme için SDK düzeyinde günlüğe kaydetmeyi etkinleştirin:

  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)
  

  Günlüğe kaydetme hakkında daha fazla bilgi için bkz. Python için Azure kitaplıklarında günlüğü yapılandırma.

• Ağ izlemeyi kullan: Ayrıntılı hata ayıklama için ağ düzeyinde izlemeyi etkinleştirin:

  Önemli

  HTTP günlüğü, üst bilgilerdeki hesap anahtarları ve diğer kimlik bilgileri gibi hassas bilgileri içerebilir. Güvenliğin tehlikeye atılmasını önlemek için bu günlükleri koruduğunızdan emin olun.

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

Zaman uyumsuz programlama için dikkat edilmesi gereken özel noktalar

Zaman uyumsuz istemciler kullandığınızda, hata yönetimi özel dikkat gerektirir.

• Temel zaman uyumsuz hata işleme

  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
  

• İptalleri işleme

  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
  

• Eşzamanlı hata işleme

  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
  

En iyi yöntemlerin özeti

Python uygulamaları için Azure SDK'da etkili hata işleme için şunları gerçekleştirmeniz gerekir:

• Hataları tahmin edin: Bulut uygulamalarının kısmi hataları düzgün bir şekilde beklemesi ve işlemesi gerekir.
• Özel durum işlemeyi kullanın: Genel AzureError işlemeye geri dönmeden önce ClientAuthenticationError ve ResourceNotFoundError gibi belirli özel durumları yakalayın.
• Akıllı yeniden deneme mantığını uygulama: Yerleşik yeniden deneme ilkelerini kullanın veya gereksinimlerinize göre özelleştirin. Tüm hataların yeniden denemeleri tetiklememesi gerektiğini unutmayın.
• Tanılama bilgilerini yakalama: Etkili sorun giderme için istek kimliği, hata kodları ve zaman damgalarını her zaman günlüğe kaydedin.
• Anlamlı kullanıcı geri bildirimi sağlayın: Destek için teknik ayrıntıları korurken teknik hataları kullanıcı dostu iletilere dönüştürün.
• Test hatası senaryoları: Uygulamanızın hata koşullarında doğru şekilde davrandığından emin olmak için test kapsamınıza hata işlemeyi ekleyin.

İlgili içerik

• Azure Core istisnalar modülü referans dokümanını gözden geçirin.
• Kimlik doğrulaması ve yetkilendirme sorunlarını giderme hakkında bilgi edinin.
• Kapsamlı uygulama izleme için Azure İzleyici OpenTelemetry'yi keşfedin.