Algemene antwoordtypen in de Azure SDK voor Python begrijpen

De Azure SDK voor Python abstraheert aanroepen naar het onderliggende Communicatieprotocol van de Azure-service, ongeacht of dat protocol HTTP of AMQP is (dat wordt gebruikt voor berichten-SDK's zoals ServiceBus en EventHubs). Als u bijvoorbeeld een van de bibliotheken gebruikt die gebruikmaakt van HTTP, maakt de Azure SDK voor Python HTTP-aanvragen en ontvangt http-antwoorden onder de schermen. De SDK abstraheert deze complexiteit, zodat u kunt werken met intuïtieve Python-objecten in plaats van onbewerkte HTTP-antwoorden of JSON-nettoladingen.

Informatie over de typen objecten die u ontvangt van SDK-bewerkingen is essentieel voor het schrijven van effectieve Azure-toepassingen. In dit artikel worden de algemene antwoordtypen beschreven die u tegenkomt en hoe deze betrekking hebben op de onderliggende HTTP-communicatie.

Opmerking

In dit artikel wordt alleen het HTTP-scenario onderzocht, niet het AMQP-scenario.

Gedeserialiseerde Python-objecten

De Azure SDK voor Python geeft prioriteit aan de productiviteit van ontwikkelaars door sterk getypte Python-objecten te retourneren uit servicebewerkingen. In plaats van JSON te parseren of HTTP-statuscodes rechtstreeks te verwerken, werkt u met resourcemodellen die Azure-resources vertegenwoordigen als Python-objecten.

Wanneer u bijvoorbeeld een blob ophaalt uit Azure Storage, ontvangt u een BlobProperties object met kenmerken zoals name, sizeen last_modified in plaats van een onbewerkte JSON-woordenlijst:

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

Waar de gegevens vandaan komen

Inzicht in de gegevensstroom helpt u te waarderen wat de SDK achter de schermen doet:

  • Uw code roept een SDK-methode aan: U roept een methode aan, zoals get_blob_properties().
  • Met de SDK wordt een HTTP-aanvraag gemaakt: De SDK bouwt de juiste HTTP-aanvraag met headers, verificatie en queryparameters.
  • De Azure-service reageert: De service retourneert een HTTP-antwoord, meestal met een JSON-nettolading in de hoofdtekst van het antwoord.
  • De SDK verwerkt het antwoord: De SDK:
    • Controleert de HTTP-statuscode.
    • Parseert de hoofdtekst van het antwoord (meestal JSON).
    • Valideert de gegevens op basis van verwachte schema's.
    • Hiermee worden de gegevens toegewezen aan Python-modelobjecten.
  • Uw code ontvangt Python-objecten: U werkt met de gedeserialiseerde objecten, niet met onbewerkte HTTP-gegevens.

Met deze abstractie kunt u zich richten op uw toepassingslogica in plaats van http-protocoldetails.

Algemene antwoordtypen

De Azure SDK voor Python maakt gebruik van verschillende standaardantwoordtypen voor alle services. Als u deze typen begrijpt, kunt u effectief met elke Azure-service werken.

Resourcemodellen

De meeste SDK-bewerkingen retourneren resourcemodellen. Deze Python-objecten vertegenwoordigen Azure-resources. De modellen zijn servicespecifiek, maar volgen consistente patronen:

# 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 voor verzamelingsresultaten

Wanneer de SDK resources opsomt, retourneert het objecten ItemPaged die paginering transparant verwerken.

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

Toegang tot het onbewerkte HTTP-antwoord

Hoewel de abstracties op hoog niveau van de SDK aan de meeste behoeften voldoen, hebt u soms toegang nodig tot het onderliggende HTTP-antwoord. Veelvoorkomende scenario's zijn:

  • Foutopsporing van mislukte aanvragen.
  • Toegang tot aangepaste antwoordheaders.
  • Aangepaste logica voor opnieuw proberen implementeren.
  • Werken met niet-standaard antwoordindelingen.

De meeste SDK-methoden accepteren een 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 en iterators

Azure-services retourneren vaak grote verzamelingen resources. De SDK gebruikt ItemPaged om deze verzamelingen efficiënt te verwerken zonder alles tegelijk in het geheugen te laden.

Automatische paginering

Met de SDK worden automatisch nieuwe pagina's opgehaald terwijl u het volgende doet:

# 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

Expliciet met pagina's werken

U kunt ook rechtstreeks met pagina's werken wanneer dat nodig is:

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)

Paginaformaat beheren

Veel lijstbewerkingen accepteren een results_per_page parameter:

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

Sommige methoden voor sommige Azure-services hebben andere mechanismen voor het beheren van het paginaformaat. Azure Key Vault en Azure Search gebruiken bijvoorbeeld de top kwarg om de resultaten per aanroep te beperken. Zie de search() voor een voorbeeld dat gebruikmaakt van de Azure Search-methode.

Speciaal geval: Langlopende bewerkingen en pollers

Sommige Azure-bewerkingen kunnen niet onmiddellijk worden voltooid. Voorbeelden zijn:

  • Virtuele machines maken of verwijderen.
  • Azure Resource Manager-sjablonen implementeren.
  • Het trainen van machine learning-modellen.
  • Grote blobs kopiëren.

Met deze bewerkingen worden pollerobjecten geretourneerd waarmee de voortgang van de bewerking wordt bijgehouden.

Werken met pollers

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 pollers

Wanneer u asynchrone/wachtpatronen gebruikt, werkt u met 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()

Voorbeeld van polling-objecten voor langlopende bewerkingen: Virtuele machines

Het implementeren van virtuele machines is een voorbeeld van een bewerking die tijd kost om te voltooien en hiermee omgaat door pollerobjecten te retourneren (LROPoller voor synchrone code, AsyncLROPoller voor asynchrone 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")

Antwoord van Access voor gepaginade resultaten

Voor gepaginade resultaten gebruikt u de by_page() methode met 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)

Beste praktijken

  • Geef de voorkeur aan abstracties op hoog niveau.

  • Werk waar mogelijk met de resourcemodellen van de SDK in plaats van onbewerkte antwoorden.

  • Vermijd toegang tot een methode die voorafgaat door een onderstrepingsteken (_). Deze methoden zijn standaard privé in Python. Er zijn geen garanties voor problemen zoals wijzigingen die fouten veroorzaken in vergelijking met openbare API's:

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

  • Paginering op de juiste manier verwerken. Itereer altijd over gepagineerde resultaten in plaats van deze te converteren naar een lijst.

    # 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

  • Gebruik poller.result() voor langlopende bewerkingen. Gebruik altijd de result() methode om ervoor te zorgen dat bewerkingen zijn voltooid:

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

  • Toegang tot onbewerkte antwoorden alleen wanneer dat nodig is. Gebruik onbewerkte responstoegang spaarzaam en alleen voor specifieke vereisten:

    # 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
  • Laatst bijgewerkt om