Python용 Azure SDK에서 생성된 오류 처리

신뢰할 수 있는 클라우드 애플리케이션을 빌드하려면 단순히 기능을 구현하는 것 이상이 필요합니다. 또한 강력한 오류 처리 전략도 요구합니다. 분산 시스템 및 클라우드 서비스를 사용하는 경우 애플리케이션은 오류 시나리오를 정상적으로 처리할 준비가 되어 있어야 합니다.

Python용 Azure SDK는 개발자가 복원력 있는 애플리케이션을 빌드할 수 있도록 설계된 포괄적인 오류 모델을 제공합니다. 이 오류 모델을 이해하는 것은 다음을 위해 중요합니다.

• 일반적인 오류 시나리오를 예측하고 처리하여 애플리케이션 안정성을 향상합니다.
• 의미 있는 오류 메시지와 정상적인 성능 저하를 통해 사용자 환경을 향상합니다.
• 관련 진단 정보를 캡처하고 로깅하여 문제 해결을 간소화합니다.

이 문서에서는 Python의 오류 아키텍처용 Azure SDK를 살펴보고 애플리케이션에서 효과적인 오류 처리를 구현하기 위한 실질적인 지침을 제공합니다.

Python용 Azure SDK에서 오류를 모델하는 방법

Python용 Azure SDK는 일반 및 특정 오류 처리 기능을 제공하는 계층적 예외 모델을 사용합니다. 이 모델의 핵심은 AzureErrorAzure 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

권한 부여 오류(일반적으로 HttpResponseError403 상태 포함)는 권한이 부족할 때 발생합니다.

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

오류 처리 모범 사례

• 특정 예외 처리 사용: 일반적인 예외로 돌아가기 전에 항상 특정 예외를 catch합니다.

  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 라이브러리에서 로깅 구성을 참조하세요.

• 네트워크 추적 사용: 딥 디버깅의 경우 네트워크 수준 추적을 사용하도록 설정합니다.

  중요합니다

  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
  

모범 사례 요약

Python 애플리케이션용 Azure SDK에서 효과적인 오류 처리를 수행하려면 다음을 수행해야 합니다.

• 오류를 예상합니다. 클라우드 애플리케이션은 부분 오류를 정상적으로 예상하고 처리해야 합니다.
• 특정 예외를 처리하세요: ResourceNotFoundError 및 ClientAuthenticationError와 같은 특정 예외들을 먼저 catch한 후, 일반적인 AzureError 처리로 넘어갑니다.
• 스마트 재시도 논리 구현: 기본 제공 재시도 정책을 사용하거나 필요에 따라 사용자 지정합니다. 모든 오류가 재시도를 트리거하는 것은 아닙니다.
• 진단 정보 캡처: 효과적인 문제 해결을 위해 항상 요청 ID, 오류 코드 및 타임스탬프를 기록합니다.
• 의미 있는 사용자 피드백을 제공합니다 . 기술 오류를 사용자 친화적인 메시지로 변환하면서 지원을 위한 기술 세부 정보를 유지합니다.
• 테스트 오류 시나리오: 테스트 검사에 오류 처리를 포함시켜 애플리케이션이 오류 조건에서 올바르게 작동하는지 확인합니다.

관련 콘텐츠

• Azure Core 예외 모듈 참조를 검토합니다.
• 인증 및 권한 부여 문제 해결에 대해 알아봅니다.
• 포괄적인 애플리케이션 모니터링을 위해 Azure Monitor OpenTelemetry 를 탐색합니다.