處理由 Azure SDK for Python 產生的錯誤

打造可靠的雲端應用程式不僅需要實作功能。同時也需要強而有力的錯誤處理策略。當你使用分散式系統和雲端服務時，你的應用程式必須準備好優雅地處理故障情境。

適用於 Python 的 Azure SDK 提供完整的錯誤模型，旨在協助開發人員建置復原應用程式。了解此錯誤模型對於以下方面至關重要：

透過預見並處理常見故障情境來提升應用程式的可靠性。
透過有意義的錯誤訊息與優雅的降級，提升使用者體驗。
透過擷取並記錄相關診斷資訊，簡化故障排除流程。

本文探討適用於 Python 的 Azure SDK 錯誤架構，並提供在應用程式中實作有效錯誤處理的實用指引。

適用於 Python 的 Azure SDK 如何對錯誤進行模型化

Python 版 Azure SDK 採用階層式例外模型，提供一般且特定的錯誤處理能力。此模型的核心是 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 無法驗證您的身分時，就會發生驗證失敗：

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

授權錯誤（通常狀態為 403 的 HttpResponseError）發生於您缺乏權限時：

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 程式庫中設定記錄。

使用網路追蹤： 若要進行深度除錯，請啟用網路層級追蹤：

這很重要

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 中，有效的錯誤處理需要：

預見失敗： 雲端應用程式必須能優雅地預期並處理部分故障。
使用特定例外處理：在回歸一般AzureError處理前，先捕獲如ResourceNotFoundError和ClientAuthenticationError等具體例外。
實作智慧重試邏輯： 使用內建的重試政策，或根據需求自訂。請記住，並非所有錯誤都應該觸發重試。
擷取診斷資訊： 務必記錄請求ID、錯誤代碼和時間戳記，以便有效排查。
提供有意義的用戶回饋： 將技術錯誤轉化為使用者友善的訊息，同時保留技術細節以供支援使用。
測試錯誤情境： 在測試覆蓋中包含錯誤處理，確保應用程式在故障條件下能正常運作。

請參考 Azure Core 例外模組參考資料。
了解如何排除認證與授權問題。
探索 Azure Monitor OpenTelemetry，提供全面的應用監控。

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-12-09