Общие сведения о типах ответов в пакете SDK Azure для Python

Пакет SDK Azure для Python абстрагирует вызовы базового протокола связи службы Azure, независимо от того, является ли этот протокол HTTP или AMQP (который используется для пакетов SDK для обмена сообщениями, таких как ServiceBus и EventHubs). Например, если вы используете одну из библиотек, использующих HTTP, пакет SDK Azure для Python выполняет HTTP-запросы и получает HTTP-ответы под капотом. Пакет SDK абстрагирует эту сложность, чтобы работать с интуитивно понятными объектами Python вместо необработанных ответов HTTP или полезных данных JSON.

Понимание типов объектов, полученных из операций ПАКЕТА SDK, важно для написания эффективных приложений Azure. В этой статье описаны распространенные типы ответов, с которыми вы сталкиваетесь, и как они связаны с базовым обменом данными через HTTP.

Замечание

В этой статье рассматривается только сценарий HTTP, а не сценарий AMQP.

Десериализированные объекты Python

Пакет SDK Azure для Python определяет производительность разработчика, возвращая строго типизированные объекты Python из операций службы. Вместо анализа json или обработки кодов состояния HTTP вы работаете с моделями ресурсов, представляющими ресурсы Azure как объекты Python.

Например, при получении большого двоичного объекта из службы хранилища Azure вы получаете объект BlobProperties с такими атрибутами, как name, size и last_modified, а не необработанный словарь JSON.

from azure.storage.blob import BlobServiceClient

# Connect to storage account
blob_service_client = BlobServiceClient.from_connection_string(connection_string)
container_client = blob_service_client.get_container_client("mycontainer")

# Get blob properties - returns a BlobProperties object
blob_client = container_client.get_blob_client("myblob.txt")
properties = blob_client.get_blob_properties()

# Access properties as Python attributes
print(f"Blob name: {properties.name}")
print(f"Blob size: {properties.size} bytes")
print(f"Last modified: {properties.last_modified}")

Откуда приходят данные

Понимание потока данных помогает оценить, что делает пакет SDK за кулисами:

• Код вызывает метод SDK: Вызывается такой метод get_blob_properties().
• Пакет SDK создает HTTP-запрос: Пакет SDK создает соответствующий HTTP-запрос с заголовками, проверкой подлинности и параметрами запроса.
• Служба Azure отвечает: Служба возвращает HTTP-ответ, обычно с полезными данными JSON в тексте ответа.
• Пакет SDK обрабатывает ответ: Пакет SDK:
  • Проверяет код состояния HTTP.
  • Анализирует текст ответа (обычно JSON).
  • Проверяет данные на основе ожидаемых схем.
  • Сопоставляет данные с объектами модели Python.
• Код получает объекты Python: Вы работаете с десериализированными объектами, а не необработанными ДАННЫМи HTTP.

Эта абстракция позволяет сосредоточиться на логике приложения, а не на деталях протокола HTTP.

Распространенные типы ответов

Пакет SDK Azure для Python использует несколько стандартных типов ответов во всех службах. Общие сведения об этих типах помогают эффективно работать с любой службой Azure.

Модели ресурсов

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

# Azure Key Vault example
from azure.keyvault.secrets import SecretClient

secret_client = SecretClient(vault_url=vault_url, credential=credential)
secret = secret_client.get_secret("mysecret")  # Returns KeyVaultSecret

print(f"Secret name: {secret.name}")
print(f"Secret value: {secret.value}")
print(f"Secret version: {secret.properties.version}")

# Azure Cosmos DB example
from azure.cosmos import CosmosClient

cosmos_client = CosmosClient(url=cosmos_url, credential=credential)
database = cosmos_client.get_database_client("mydatabase")
container = database.get_container_client("mycontainer")
item = container.read_item(item="item-id", partition_key="partition-value")  # Returns dict

print(f"Item ID: {item['id']}")

ItemPaged для результатов сбора

Когда SDK перечисляет ресурсы, он возвращает ItemPaged объекты, которые прозрачно обрабатывают разбиение на страницы.

from azure.storage.blob import BlobServiceClient
from azure.core.paging import ItemPaged

blob_service_client = BlobServiceClient.from_connection_string(connection_string)
container_client = blob_service_client.get_container_client("mycontainer")

# list_blobs returns ItemPaged[BlobProperties]
blobs: ItemPaged[BlobProperties] = container_client.list_blobs()

# Iterate naturally - SDK handles pagination
for blob in blobs:
    print(f"Blob: {blob.name}, Size: {blob.size}")

Доступ к необработанному HTTP-ответу

Хотя высокоуровневые абстракции пакета SDK соответствуют большинству потребностей, иногда требуется доступ к базовому HTTP-ответу. Ниже приведены распространенные сценарии.

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

Большинство методов ПАКЕТА SDK принимают raw_response_hook параметр:

from azure.keyvault.secrets import SecretClient

secret_client = SecretClient(vault_url=vault_url, credential=credential)

def inspect_response(response):
    # Access the raw HTTP response
    print(f"Request URL: {response.http_request.url}")
    print(f"Status code: {response.http_response.status_code}")
    print(f"Response headers: {dict(response.http_response.headers)}")
    
    # Access custom headers
    request_id = response.http_response.headers.get('x-ms-request-id')
    print(f"Request ID: {request_id}")
    
    # Must return the response
    return response

# Hook is called before deserialization
secret = secret_client.get_secret("mysecret", raw_response_hook=inspect_response)

Разбиение по страницам и итераторам

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

Автоматическое разбиение на страницы

Пакет SDK автоматически извлекает новые страницы при итерации:

# List all blobs - could be thousands
blobs = container_client.list_blobs()

# SDK fetches pages as needed during iteration
for blob in blobs:
    process_blob(blob)  # Pages loaded on-demand

Работа с страницами явным образом

Вы также можете работать с страницами непосредственно при необходимости:

blobs = container_client.list_blobs()

# Process by page
for page in blobs.by_page():
    print(f"Processing page with {len(list(page))} items")
    for blob in page:
        process_blob(blob)

Управление размером страницы

Многие операции списка принимают results_per_page параметр:

# Fetch 100 items per page instead of the default
blobs = container_client.list_blobs(results_per_page=100)

Некоторые методы для некоторых служб Azure имеют другие механизмы управления размером страницы. Например, Azure Key Vault и Поиск Azure используют top kwarg, чтобы ограничить результаты для каждого вызова. Пример использования метода поиска search() Azure см. в исходном коде.

Особый случай: длительные операции и опросы

Некоторые операции Azure не могут завершиться немедленно. Вот некоторые примеры.

• Создание или удаление виртуальных машин.
• Развертывание шаблонов Azure Resource Manager.
• Обучение моделей машинного обучения.
• Копирование больших BLOB-объектов.

Эти операции возвращают объекты опроса, отслеживающие ход выполнения операции.

Работа с опросами

from azure.mgmt.storage import StorageManagementClient

storage_client = StorageManagementClient(credential, subscription_id)

# Start storage account creation
poller = storage_client.storage_accounts.begin_create(
    resource_group_name="myresourcegroup",
    account_name="mystorageaccount",
    parameters=storage_parameters
)

# Option 1: Wait for completion (blocking)
storage_account = poller.result()

# Option 2: Check status periodically
while not poller.done():
    print(f"Status: {poller.status()}")
    time.sleep(5)

storage_account = poller.result()

Асинхронные опросы

При использовании шаблонов async/await вы работаете с AsyncLROPoller.

from azure.storage.blob.aio import BlobServiceClient

async with BlobServiceClient.from_connection_string(connection_string) as client:
    container_client = client.get_container_client("mycontainer")
    
    # Start async copy operation
    blob_client = container_client.get_blob_client("large-blob.vhd")
    poller = await blob_client.begin_copy_from_url(source_url)
    
    # Wait for async completion
    copy_properties = await poller.result()

Пример опроса объектов для продолжительных операций: виртуальные машины

Развертывание виртуальных машин является примером операции, которая занимает время для завершения и предполагает возврат опросных объектов (LROPoller для синхронного кода, AsyncLROPoller для асинхронного кода).

from azure.mgmt.compute import ComputeManagementClient
from azure.core.polling import LROPoller

compute_client = ComputeManagementClient(credential, subscription_id)

# Start VM creation - returns immediately with a poller
poller: LROPoller = compute_client.virtual_machines.begin_create_or_update(
    resource_group_name="myresourcegroup",
    vm_name="myvm",
    parameters=vm_parameters
)

# Wait for completion and get the result
vm = poller.result()  # Blocks until operation completes
print(f"VM {vm.name} provisioned successfully")

Запрос на доступ к странично отображаемым результатам

Для страничных результатов используйте метод со by_page()следующими параметрамиraw_response_hook:

def page_response_hook(response):
    continuation_token = response.http_response.headers.get('x-ms-continuation')
    print(f"Continuation token: {continuation_token}")
    return response

blobs = container_client.list_blobs()
for page in blobs.by_page(raw_response_hook=page_response_hook):
    for blob in page:
        print(blob.name)

Лучшие практики

• Предпочитайте высокоуровневые абстракции.

• По возможности работайте с моделями ресурсов пакета SDK, а не необработанными ответами.

• Избегайте вызова методов, начинающихся с подчеркивания (_). По соглашению эти методы являются частными в Python. Нет никаких гарантий в отношении таких проблем, как изменения, нарушающие совместимость, в общедоступных API.

  # Preferred: Work with typed objects
  secret = secret_client.get_secret("mysecret")
  if secret.properties.enabled:
      use_secret(secret.value)
  
  # Avoid: Manual JSON parsing (unless necessary) ...
  # AND avoid accessing any objects or methods that start with `_`
  response = secret_client._client.get(...)  # Don't access internal clients
  data = json.loads(response.text)
  if data['attributes']['enabled']:
      use_secret(data['value'])
  

• Правильно обработайте разбиение на страницы. Всегда итерировать по страницам результаты вместо преобразования в список:

  # Good: Memory-efficient iteration
  for blob in container_client.list_blobs():
      process_blob(blob)
  
  # Avoid: Loading everything into memory
  all_blobs = list(container_client.list_blobs())  # Could consume excessive memory
  

• Используйте poller.result() для длительных операций. Всегда используйте метод result(), чтобы обеспечить успешное выполнение операций.

  # Correct: Wait for operation completion
  poller = compute_client.virtual_machines.begin_delete(
      resource_group_name="myresourcegroup",
      vm_name="myvm"
  )
  poller.result()  # Ensures deletion completes
  print("VM deleted successfully")
  
  # Wrong: Assuming immediate completion
  poller = compute_client.virtual_machines.begin_delete(...)
  print("VM deleted successfully")  # Deletion might still be in progress!
  

• Доступ к необработанным ответам только при необходимости. Используйте доступ к необработанным ответам экономно и только при конкретных требованиях.

  # Good use case: Debugging or logging
  def log_request_id(response):
      request_id = response.http_response.headers.get('x-ms-request-id')
      logger.info(f"Operation request ID: {request_id}")
      return response
  
  blob_client.upload_blob(data, raw_response_hook=log_request_id)
  
  # Good use case: Custom error handling
  def check_custom_header(response):
      if response.http_response.headers.get('x-custom-error'):
          raise CustomApplicationError("Custom error condition detected")
      return response