Azure SDK for Python によって生成されたエラーを処理する

信頼性の高いクラウド アプリケーションを構築するには、単なる機能の実装以上のものが必要です。 また、堅牢なエラー処理戦略も必要です。 分散システムとクラウド サービスを使用する場合は、障害シナリオを適切に処理できるようにアプリケーションを準備する必要があります。

Azure SDK for Python は、開発者が回復性のあるアプリケーションを構築するのに役立つように設計された包括的なエラー モデルを提供します。 このエラー モデルを理解することは、次の場合に非常に重要です。

• 一般的な障害シナリオを予測して処理することで、アプリケーションの信頼性を向上させる。
• 意味のあるエラー メッセージと正常な低下を通じてユーザー エクスペリエンスを強化します。
• 関連する診断情報をキャプチャしてログに記録することで、トラブルシューティングを簡略化します。

この記事では、Azure SDK for Python のエラー アーキテクチャについて説明し、アプリケーションで効果的なエラー処理を実装するための実用的なガイダンスを提供します。

Azure SDK for Python でエラーがどのようにモデル化されるか

Azure SDK for Python では、一般的で特定のエラー処理機能を提供する階層的な例外モデルを使用します。 このモデルの中核となるのは AzureErrorであり、Azure SDK に関連するすべてのエラーの基本例外クラスとして機能します。

例外階層

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

主な例外の種類

エラー	Description
AzureError	すべての Azure SDK エラーの基本例外クラス。 Azure 関連のエラーを処理する必要がある場合は、この例外をキャッチオールとして使用します。
ClientAuthenticationError	認証が失敗したときに発生します。 一般的な原因としては、無効な資格情報、期限切れのトークン、正しく構成されていない認証設定などがあります。
ResourceNotFoundError	存在しないリソースにアクセスしようとしたときに発生します。 通常、この例外は HTTP 404 応答に対応します。
ResourceExistsError	既に存在するリソースを作成しようとしたときに発生します。 この例外は、誤って上書きされるのを防ぐのに役立ちます。
ServiceRequestError	SDK がサービスに要求を送信できない場合に発生します。 一般的な原因としては、ネットワーク接続の問題、ドメイン ネーム システムの解決エラー、無効なサービス エンドポイントなどがあります。
ServiceResponseError	SDK で処理できない予期しない応答がサービスから返されたときに発生します。
HttpResponseError	HTTP エラー応答 (4xx および 5xx 状態コード) に対して発生しました。 この例外は、基になる HTTP 応答の詳細へのアクセスを提供します。

一般的なエラー シナリオ

一般的なエラー シナリオを理解することは、状況ごとに適切な処理戦略を実装するのに役立ちます。

認証と承認のエラー

認証エラーは、SDK が ID を確認できない場合に発生します。

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

承認エラー (通常はHttpResponseError状態で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

リソース エラー

不足しているリソースを適切に処理します。

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"

重複するリソースの作成を防止する:

from azure.core.exceptions import ResourceExistsError

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

サーバー エラー

サーバー側の障害を適切に処理します。

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

エラー処理のベスト プラクティス

• 特定の例外処理を使用します。 一般的な例外にフォールバックする前に、常に特定の例外をキャッチしてください。

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

• 適切な再試行戦略を実装します。 再試行を保証するエラーもあれば、再試行しないエラーもあります。

  次の操作を再試行しないでください。

  • 401 未承認 (認証エラー)
  • 403 禁止 (承認エラー)
  • 400 無効な要求 (クライアント エラー)
  • 404 見つかりません (リソースが表示される場合を除く)

  次の点で再試行することを検討してください。

  • 408 要求タイムアウト
  • 429 要求が多すぎます (適切なバックオフあり)
  • 500 内部サーバー エラー
  • 502 無効なゲートウェイ
  • 503 サービスを利用できない
  • 504 ゲートウェイ タイムアウト

• 意味のあるエラー情報を抽出する

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

再試行ポリシーと回復性

Azure SDK には、一時的な障害を自動的に処理する再試行メカニズムが組み込まれています。

既定の再試行動作

ほとんどの Azure SDK クライアントには、次のような既定の再試行ポリシーが含まれています。

• 接続エラーと特定の HTTP 状態コードを再試行します。
• ジッターを用いた指数バックオフを使用します。
• 再試行回数を制限します。

再試行ポリシーをカスタマイズする

既定の動作がユース ケースに合わない場合は、再試行ポリシーをカスタマイズできます。

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
)

カスタム ループによるネットワーク エラーとタイムアウト エラーの処理を回避する

独自のカスタム ロジックを実装する前に、ネットワーク エラーとタイムアウト エラーに組み込みの再試行を使用してみてください。

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

サーキット ブレーカー パターンを実装する

重要な操作の場合は、サーキット ブレーカー パターンの実装を検討してください。

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

エラー メッセージとコードについて

Azure サービスは、貴重なデバッグ情報を提供する構造化されたエラー応答を返します。

• エラー応答の解析

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

• 診断情報をキャプチャします。 トラブルシューティングのために、常にキー診断情報をキャプチャします。

  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
  

• ログと診断: SDK レベルのログ記録を有効にして、詳細なトラブルシューティングを行います。

  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)
  

  ログ記録の詳細については、「 Python 用 Azure ライブラリでのログ記録の構成」を参照してください。

• ネットワーク トレースを使用する: 詳細デバッグの場合は、ネットワーク レベルのトレースを有効にします。

  Important

  HTTP ログには、ヘッダーやその他の資格情報にアカウント キーなどの機密情報を含めることができます。 セキュリティを損なわないように、これらのログを必ず保護してください。

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

非同期プログラミングに関する特別な考慮事項

非同期クライアントを使用する場合、エラー処理には特別な注意が必要です。

• 基本的な非同期エラー処理

  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
  

• 取り消しの処理

  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
  

• 同時実行エラー処理

  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
  

ベスト プラクティスの概要

Azure SDK for Python アプリケーションで有効なエラー処理を行うには、次の操作が必要です。

• エラーを予測する: クラウド アプリケーションでは、部分的な障害を適切に想定して処理する必要があります。
• 特定の例外処理を使用します。一般的なResourceNotFoundError処理にフォールバックする前に、ClientAuthenticationErrorやAzureErrorなどの特定の例外をキャッチします。
• スマート再試行ロジックを実装します。 組み込みの再試行ポリシーを使用するか、ニーズに基づいてカスタマイズします。 すべてのエラーで再試行がトリガーされるわけではないことに注意してください。
• 診断情報をキャプチャします。 効果的なトラブルシューティングを行うには、要求 ID、エラー コード、タイムスタンプを常にログに記録します。
• 意味のあるユーザー フィードバックを提供します。 サポートの技術的な詳細を保持しながら、技術的なエラーをわかりやすいメッセージに変換します。
• テスト エラーシナリオ: テスト カバレッジにエラー処理を含め、障害条件下でアプリケーションが正しく動作することを確認します。

関連コンテンツ

• Azure Core 例外モジュール リファレンスを確認します。
• 認証と承認に関する問題のトラブルシューティングについて説明します。
• Azure Monitor OpenTelemetry を調べて、包括的なアプリケーション監視を行います。