Grundlegendes zu allgemeinen Antworttypen im Azure SDK für Python

Das Azure SDK für Python abstrahiert die Aufrufe an das zugrunde liegende Kommunikationsprotokoll von Azure-Diensten, unabhängig davon, ob dieses Protokoll HTTP oder AMQP ist (das für Messaging-SDKs wie ServiceBus und EventHubs verwendet wird). Wenn Sie beispielsweise eine der Bibliotheken verwenden, die HTTP verwenden, sendet das Azure SDK für Python HTTP-Anforderungen und empfängt HTTP-Antworten unter der Haube. Das SDK abstrahiert diese Komplexität, sodass Sie mit intuitiven Python-Objekten anstelle von unformatierten HTTP-Antworten oder JSON-Nutzlasten arbeiten können.

Das Verständnis der Objekttypen, die Sie von SDK-Vorgängen erhalten, ist für das Schreiben effektiver Azure-Anwendungen unerlässlich. In diesem Artikel werden die häufig auftretenden Antworttypen und deren Beziehung zur zugrunde liegenden HTTP-Kommunikation erläutert.

Hinweis

In diesem Artikel wird nur das HTTP-Szenario und nicht das AMQP-Szenario untersucht.

Deserialisierte Python-Objekte

Das Azure SDK für Python priorisiert die Produktivität der Entwickler, indem stark typisierte Python-Objekte aus Dienstvorgängen zurückgegeben werden. Anstatt JSON zu analysieren oder HTTP-Statuscodes direkt zu verarbeiten, arbeiten Sie mit Ressourcenmodellen, die Azure-Ressourcen als Python-Objekte darstellen.

Wenn Sie beispielsweise ein Blob aus Azure Storage abrufen, erhalten Sie ein BlobProperties Objekt mit Attributen wie name, size und last_modified, statt einem unformatierten JSON-Wörterbuch.

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

Woher die Daten stammen

Das Verständnis des Datenflusses hilft Ihnen zu schätzen, was das SDK hinter den Kulissen tut:

  • Ihr Code ruft eine SDK-Methode auf: Sie rufen eine Methode wie get_blob_properties().
  • Das SDK erstellt eine HTTP-Anforderung: Das SDK erstellt die entsprechende HTTP-Anforderung mit Headern, Authentifizierungs- und Abfrageparametern.
  • Der Azure-Dienst antwortet: Der Dienst gibt eine HTTP-Antwort zurück, in der Regel mit einer JSON-Nutzlast im Antworttext.
  • Das SDK verarbeitet die Antwort: Das SDK:
    • Überprüft den HTTP-Statuscode.
    • Analysiert den Antworttext (in der Regel JSON).
    • Überprüft die Daten anhand erwarteter Schemas.
    • Ordnet die Daten den Python-Modellobjekten zu.
  • Ihr Code empfängt Python-Objekte: Sie arbeiten mit den deserialisierten Objekten, nicht mit unformatierten HTTP-Daten.

Mit dieser Abstraktion können Sie sich nicht auf HTTP-Protokolldetails, sondern auf Ihre Anwendungslogik konzentrieren.

Allgemeine Antworttypen

Das Azure SDK für Python verwendet mehrere Standardantworttypen für alle Dienste. Wenn Sie diese Typen verstehen, können Sie effektiv mit jedem Azure-Dienst arbeiten.

Ressourcenmodelle

Die meisten SDK-Vorgänge geben Ressourcenmodelle zurück. Diese Python-Objekte stellen Azure-Ressourcen dar. Die Modelle sind dienstspezifisch, folgen jedoch konsistenten Mustern:

# 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 für Sammlungsergebnisse

Wenn das SDK Ressourcen auflistet, werden Objekte zurückgegeben ItemPaged , die die Paginierung transparent behandeln:

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

Zugreifen auf die unformatierte HTTP-Antwort

Während die allgemeinen Abstraktionen des SDK den meisten Anforderungen entsprechen, benötigen Sie manchmal Zugriff auf die zugrunde liegende HTTP-Antwort. Zu den häufigen Szenarios gehören:

  • Debuggen fehlgeschlagener Anfragen.
  • Zugreifen auf benutzerdefinierte Antwortheader.
  • Implementieren von benutzerdefinierter Wiederholungslogik.
  • Arbeiten mit nicht standardmäßigen Antwortformaten.

Die meisten SDK-Methoden akzeptieren einen raw_response_hook Parameter:

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)

Paging und Iteratoren

Azure-Dienste geben häufig große Sammlungen von Ressourcen zurück. Das SDK nutzt ItemPaged zur effizienten Verwaltung dieser Auflistungen, ohne alles gleichzeitig in den Arbeitsspeicher zu laden.

Automatischer Seitenumbruch

Das SDK ruft beim Durchlaufen automatisch neue Seiten ab:

# 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

Explizites Arbeiten mit Seiten

Sie können auch bei Bedarf direkt mit Seiten arbeiten:

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)

Seitengröße steuern

Viele Listenvorgänge akzeptieren einen results_per_page Parameter:

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

Einige Methoden für einige Azure-Dienste weisen andere Mechanismen zum Steuern der Seitengröße auf. Beispielsweise verwenden Azure Key Vault und Azure Search den top Kwarg, um die Ergebnisse pro Anruf einzuschränken. Ein Beispiel, das die Azure Search-Methode search() verwendet, finden Sie im Quellcode.

Sonderfall: Lange ausgeführte Vorgänge und Poller

Einige Azure-Vorgänge können nicht sofort abgeschlossen werden. Beispiele sind:

  • Erstellen oder Löschen virtueller Computer.
  • Bereitstellen von Azure Resource Manager-Vorlagen.
  • Trainieren von Machine Learning-Modellen.
  • Kopieren großer Blobs.

Diese Vorgänge geben Abrufobjekte zurück, die den Fortschritt des Vorgangs nachverfolgen.

Arbeiten mit Umfragen

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

Asynchrone Umfragen

Wenn Sie asynchrone/await-Muster verwenden, arbeiten Sie mit 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()

Abrufen von Objekten für langfristige Vorgänge: Beispiel Virtuelle Computer

Das Bereitstellen virtueller Computer ist ein Beispiel für einen Vorgang, der Zeit benötigt und durch das Zurückgeben von Abrufobjekten gehandhabt wird (LROPoller für synchronen Code und AsyncLROPoller für asynchronen Code).

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

Zugriffsantwort für seitenseitige Ergebnisse

Verwenden Sie für seitenseitige Ergebnisse die by_page() Methode mit 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)

Bewährte Methoden

  • Bevorzugen Sie allgemeine Abstraktionen.

  • Arbeiten Sie mit den Ressourcenmodellen des SDK und nicht mit unformatierten Antworten, wenn möglich.

  • Vermeiden Sie den Zugriff auf eine Methode, die einem Unterstrich (_) vorangestellt ist. Standardmäßig sind diese Methoden in Python privat. Es gibt keine Garantien für Probleme wie das Unterbrechen von Änderungen im Vergleich zu öffentlichen APIs:

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

  • Behandeln Sie die Paginierung ordnungsgemäß. Durchlaufen Sie immer paginierte Ergebnisse, anstatt sie in eine Liste umzuwandeln.

    # 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

  • Verwenden Sie poller.result() für lang andauernde Vorgänge. Verwenden Sie immer die result() Methode, um sicherzustellen, dass Vorgänge erfolgreich abgeschlossen werden:

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

  • Greifen Sie nur bei Bedarf auf unformatierte Antworten zu. Nur für bestimmte Anforderungen können Sie den Zugriff auf unformatierte Antworten sparsam verwenden:

    # 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
  • Zuletzt aktualisiert um