Omówienie typowych typów odpowiedzi w zestawie Azure SDK dla języka Python

Zestaw Azure SDK dla języka Python abstrahuje wywołania podstawowego protokołu komunikacyjnego usługi platformy Azure, niezależnie od tego, czy jest to protokół HTTP, czy AMQP (który jest używany do zestawów SDK do obsługi komunikatów, takich jak ServiceBus i EventHubs). Jeśli na przykład używasz jednej z bibliotek używających protokołu HTTP, zestaw Azure SDK dla języka Python wysyła żądania HTTP i odbiera odpowiedzi HTTP pod maską. Zestaw SDK oddziela tę złożoność tak, aby można było pracować z intuicyjnymi obiektami języka Python zamiast nieprzetworzonych odpowiedzi HTTP lub ładunków JSON.

Zrozumienie typów obiektów otrzymywanych z operacji zestawu SDK jest niezbędne do pisania skutecznych aplikacji platformy Azure. W tym artykule opisano typowe typy odpowiedzi, które napotykasz, oraz sposób ich powiązania z podstawową komunikacją HTTP.

Uwaga / Notatka

W tym artykule przedstawiono tylko scenariusz HTTP, a nie scenariusz protokołu AMQP.

Deserializowane obiekty języka Python

Zestaw Azure SDK dla języka Python określa priorytet produktywności deweloperów, zwracając silnie typizowane obiekty języka Python z operacji usługi. Zamiast analizować kod JSON lub bezpośrednio obsługiwać kody stanu HTTP, pracujesz z modelami zasobów reprezentującymi zasoby platformy Azure jako obiekty języka Python.

Na przykład podczas pobierania obiektu blob z usługi Azure Storage otrzymasz BlobProperties obiekt z atrybutami takimi jak name, sizei last_modified zamiast nieprzetworzonego słownika 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}")

Skąd pochodzą dane

Zrozumienie przepływu danych pomaga docenić działania zestawu SDK w tle:

• Kod wywołuje metodę zestawu SDK: Wywołasz metodę, na przykład get_blob_properties().
• Zestaw SDK tworzy żądanie HTTP: Zestaw SDK tworzy odpowiednie żądanie HTTP z nagłówkami, uwierzytelnianiem i parametrami zapytania.
• Usługa platformy Azure odpowiada: Usługa zwraca odpowiedź HTTP, zazwyczaj z ładunkiem JSON w treści odpowiedzi.
• Zestaw SDK przetwarza odpowiedź: Zestaw SDK:
  • Sprawdza kod stanu HTTP.
  • Analizuje treść odpowiedzi (zazwyczaj JSON).
  • Sprawdza poprawność danych względem oczekiwanych schematów.
  • Mapuje dane na obiekty modelowe w Pythonie.
• Kod odbiera obiekty języka Python: Pracujesz z obiektami deserializacji, a nie nieprzetworzonymi danymi HTTP.

Ta abstrakcja umożliwia skoncentrowanie się na logice aplikacji, a nie na szczegółach protokołu HTTP.

Typowe typy odpowiedzi

Zestaw Azure SDK dla języka Python używa kilku standardowych typów odpowiedzi we wszystkich usługach. Zrozumienie tych typów ułatwia efektywną pracę z dowolną usługą platformy Azure.

Modele zasobów

Większość operacji zestawu SDK zwraca modele zasobów. Te obiekty języka Python reprezentują zasoby platformy Azure. Modele są specyficzne dla usługi, ale są zgodne ze spójnymi wzorcami:

# 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']}")

ElementPaged dla wyników kolekcji

Gdy SDK wyświetla listę zasobów, zwraca ItemPaged obiekty, które w sposób przejrzysty obsługują stronicowanie:

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

Uzyskiwanie dostępu do nieprzetworzonej odpowiedzi HTTP

Chociaż abstrakcje wysokiego poziomu zestawu SDK spełniają większość potrzeb, czasami potrzebujesz dostępu do bazowej odpowiedzi HTTP. Typowe scenariusze obejmują:

• Debugowanie żądań nie powiodło się.
• Uzyskiwanie dostępu do niestandardowych nagłówków odpowiedzi.
• Implementowanie niestandardowej logiki ponawiania prób.
• Praca z niestandardowymi formatami odpowiedzi.

Większość metod zestawu SDK akceptuje raw_response_hook parametr:

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)

Stronicowanie i iteratory

Usługi platformy Azure często zwracają duże kolekcje zasobów. Zestaw SDK używa ItemPaged do wydajnego obsługi tych kolekcji bez konieczności ładowania wszystkich elementów do pamięci jednocześnie.

Automatyczne stronicowanie

Zestaw SDK automatycznie pobiera nowe strony podczas iterowania:

# 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

Jawna praca ze stronami

Możesz również pracować z stronami bezpośrednio w razie potrzeby:

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)

Sterowanie rozmiarem strony

Wiele operacji listy akceptuje results_per_page parametr:

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

Niektóre metody niektórych usług platformy Azure mają inne mechanizmy kontrolowania rozmiaru strony. Na przykład usługi Azure Key Vault i Azure Search używają top kwarg, aby ograniczyć wyniki na wywołanie. Aby zapoznać się z przykładem użycia metody usługi Azure Search search() , zobacz kod źródłowy.

Szczególny przypadek: Długotrwałe operacje i sondatory

Niektóre operacje platformy Azure nie mogą być wykonywane natychmiast. Oto kilka przykładów:

• Tworzenie lub usuwanie maszyn wirtualnych.
• Wdrażanie szablonów usługi Azure Resource Manager.
• Trenowanie modeli uczenia maszynowego.
• Kopiowanie dużych blobów.

Te operacje zwracają obiekty poller, które śledzą postęp operacji.

Praca z ankieterami

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

Sondy asynchroniczne

W przypadku używania wzorców async/await pracujesz z 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()

Przykład sondowania obiektów dla długotrwałych operacji: Maszyny wirtualne

Wdrażanie maszyn wirtualnych to przykład operacji, która zajmuje trochę czasu i obsługuje ją poprzez zwrócenie obiektów poller (LROPoller dla kodu synchronicznego, AsyncLROPoller dla kodu asynchronicznego):

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

Odpowiedź dostępu dla stronicowanych wyników

W przypadku stronicowanych wyników użyj metody z poleceniem 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)

Najlepsze rozwiązania

• Preferuj abstrakcję wysokiego poziomu.

• Współpracuj z modelami zasobów zestawu SDK, a nie nieprzetworzonymi odpowiedziami, gdy jest to możliwe.

• Unikaj uzyskiwania dostępu do dowolnej metody poprzedzonej znakiem podkreślenia (_). Zgodnie z konwencją te metody są prywatne w języku Python. Nie ma gwarancji dotyczących problemów, takich jak zmiany powodujące niezgodność w porównaniu z publicznymi interfejsami 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'])
  

• Poprawnie obsłuż stronicowanie. Zawsze iteruj po wynikach stronicowanych zamiast je konwertować na listę.

  # 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
  

• Użyj metody poller.result() w przypadku długotrwałych operacji. Zawsze używaj result() metody , aby upewnić się, że operacje zakończą się pomyślnie:

  # 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!
  

• Uzyskaj dostęp do nieprzetworzonych odpowiedzi tylko wtedy, gdy jest to konieczne. Używaj dostępu do odpowiedzi nieprzetworzonej oszczędnie i tylko w przypadku określonych wymagań:

  # 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