Обработка ошибок, создаваемых пакетом SDK Azure для Python

Для создания надежных облачных приложений требуется больше, чем просто реализация функций. Он также требует надежных стратегий обработки ошибок. При работе с распределенными системами и облачными службами приложение должно быть готово к обработке сценариев сбоев.

Пакет SDK Azure для Python предоставляет комплексную модель ошибок, которая поможет разработчикам создавать устойчивые приложения. Понимание этой модели ошибок имеет решающее значение для:

Повышение надежности приложений путем прогнозирования и обработки распространенных сценариев сбоя.
Улучшение взаимодействия с пользователем с помощью значимых сообщений об ошибках и плавного ухудшения.
Упрощение устранения неполадок путем записи и ведения журнала соответствующих диагностических сведений.

В этой статье рассматриваются архитектуры ошибок Azure SDK для Python и приведены практические рекомендации по реализации эффективной обработки ошибок в приложениях.

Как пакет SDK Для Azure для моделей Python ошибок

Пакет SDK Azure для Python использует иерархическую модель исключений, которая предоставляет общие и конкретные возможности обработки ошибок. В основе этой модели лежит AzureErrorбазовый класс исключений для всех ошибок, связанных с пакетом SDK Azure.

Иерархия исключений

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

Типы исключений ключей

Ошибка	Description
`AzureError`	Базовый класс исключений для всех ошибок пакета SDK Azure. Используйте это исключение как универсальное решение, если необходимо обработать любую ошибку, связанную с 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

Ошибки авторизации (обычно 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

Политики повторных попыток и устойчивость

Пакет SDK Azure включает встроенные механизмы повторных попыток, которые обрабатывают временные сбои автоматически.

Поведение повторных попыток по умолчанию

Большинство клиентов Пакета SDK Azure включают политики повторных попыток по умолчанию, которые:

Повторите попытку при возникновении ошибок подключения и определенных кодов состояния 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)
```
Дополнительные сведения о ведении журнала см. в разделе "Настройка ведения журнала" в библиотеках Azure для Python.
Используйте трассировку сети: Для глубокой отладки включите трассировку на уровне сети:

Это важно

Ведение журнала 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 для приложений Python требуется:

Предвидеть сбои: Облачные приложения должны ожидать и обрабатывать частичные сбои.
Используйте определенную обработку исключений: Перехват конкретных исключений, таких как ResourceNotFoundError и ClientAuthenticationError прежде, чем вернуться к обычной AzureError обработке.
Реализуйте интеллектуальную логику повторных попыток: Используйте встроенные политики повторных попыток или настройте их на основе ваших потребностей. Помните, что не все ошибки должны активировать повторные попытки.
Сбор диагностических сведений: Всегда регистрируют идентификаторы запросов, коды ошибок и метки времени для эффективного устранения неполадок.
Предоставьте значимые отзывы пользователей: Преобразуйте технические ошибки в понятные сообщения при сохранении технических сведений для поддержки.
Сценарии тестирования ошибок: Включите обработку ошибок в тестовое покрытие, чтобы убедиться, что приложение работает правильно в условиях сбоя.

Ознакомьтесь со справочным руководством модуля исключений Azure Core.
Узнайте об устранении неполадок с проверкой подлинности и авторизацией.
Ознакомьтесь с Azure Monitor OpenTelemetry для комплексного мониторинга приложений.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2025-08-27