Förstå vanliga svarstyper i Azure SDK för Python

Azure SDK för Python abstraherar anrop till det underliggande Azure-tjänstkommunikationsprotokollet, oavsett om protokollet är HTTP eller AMQP (som används för meddelande-SDK:er som ServiceBus och EventHubs). Om du till exempel använder ett av biblioteken som använder HTTP gör Azure SDK för Python HTTP-begäranden och tar emot HTTP-svar under huven. SDK:t abstraherar bort den här komplexiteten så att du kan arbeta med intuitiva Python-objekt i stället för råa HTTP-svar eller JSON-nyttolaster.

Det är viktigt att förstå vilka typer av objekt som du får från SDK-åtgärder för att skriva effektiva Azure-program. Den här artikeln beskriver de vanliga svarstyper som du stöter på och hur de relaterar till den underliggande HTTP-kommunikationen.

Anmärkning

Den här artikeln undersöker endast HTTP-scenariot, inte AMQP-scenariot.

Deserialiserade Python-objekt

Azure SDK för Python prioriterar utvecklarproduktiviteten genom att returnera starkt inskrivna Python-objekt från tjänståtgärder. I stället för att parsa JSON eller hantera HTTP-statuskoder direkt arbetar du med resursmodeller som representerar Azure-resurser som Python-objekt.

När du till exempel hämtar en blob från Azure Storage får du ett BlobProperties objekt med attribut som name, sizeoch last_modified i stället för en rå JSON-ordlista:

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

Var data kommer från

Att förstå dataflödet hjälper dig att uppskatta vad SDK:et gör i bakgrunden:

Koden anropar en SDK-metod: Du anropar en metod som get_blob_properties().
SDK:t skapar en HTTP-begäran: SDK:t skapar lämplig HTTP-begäran med rubriker, autentisering och frågeparametrar.
Azure-tjänsten svarar: Tjänsten returnerar ett HTTP-svar, vanligtvis med en JSON-nyttolast i svarstexten.
SDK bearbetar svaret: SDK:et:
- Kontrollerar HTTP-statuskoden.
- Parsar svarstexten (vanligtvis JSON).
- Verifierar data mot förväntade scheman.
- Mappar data till Python-modellobjekt.
Koden tar emot Python-objekt: Du arbetar med deserialiserade objekt, inte råa HTTP-data.

Med den här abstraktionen kan du fokusera på programlogik i stället för HTTP-protokollinformation.

Vanliga svarstyper

Azure SDK för Python använder flera standardsvarstyper för alla tjänster. Genom att förstå dessa typer kan du arbeta effektivt med alla Azure-tjänster.

Resursmodeller

De flesta SDK-åtgärder returnerar resursmodeller. Dessa Python-objekt representerar Azure-resurser. Modellerna är tjänstspecifika men följer konsekventa mönster:

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

När SDK:et listar resurser returneras ItemPaged objekt som hanterar paginering transparent.

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

Få åtkomst till HTTP-råsvaret

Även om SDK:s abstraktioner på hög nivå uppfyller de flesta behov behöver du ibland åtkomst till det underliggande HTTP-svaret. Vanliga scenarier är:

Felsökning av misslyckade begäranden.
Åtkomst till anpassade HTTP-svarshuvuden.
Implementera anpassad logik för återförsök.
Arbeta med icke-standardsvarsformat.

De flesta SDK-metoder accepterar en 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)

Paginering och iteratorer

Azure-tjänster returnerar ofta stora samlingar med resurser. SDK:et använder ItemPaged för att hantera dessa samlingar effektivt utan att läsa in allt i minnet på en gång.

Automatisk paginering

SDK:n hämtar automatiskt nya sidor när du itererar:

# 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

Arbeta uttryckligen med sidor

Du kan också arbeta med sidor direkt när det behövs:

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)

Kontrollsidans storlek

Många liståtgärder accepterar en results_per_page parameter:

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

Vissa metoder för vissa Azure-tjänster har andra mekanismer för att kontrollera sidstorleken. För exempel använder Azure Key Vault och Azure Search top kwarg för att begränsa resultaten per samtal. Ett exempel som använder Azure Search-metoden search() finns i källkoden.

Specialfall: Långvariga operationer och pollers

Vissa Azure-åtgärder kan inte slutföras omedelbart. Exempel är:

Skapa eller ta bort virtuella datorer.
Distribuera Azure Resource Manager-mallar.
Träna maskininlärningsmodeller.
Kopierar stora blobar.

Dessa åtgärder returnerar pollerobjekt som spårar åtgärdens förlopp.

Arbeta med polling-verktyg

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

Asynkrona pollare

När du använder asynkrona/väntande mönster arbetar du med 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()

Exempel på avsökningsobjekt för långvariga åtgärder: Virtuella datorer

Att distribuera virtuella datorer är ett exempel på en åtgärd som tar tid att slutföra och hanteras genom att returnera pollerobjekt (LROPoller för synkron kod, AsyncLROPoller för asynkron kod).

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

Åtkomstsvar för sidsidiga resultat

För sidsidiga resultat använder du by_page() metoden med 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)

Metodtips

Föredrar abstraktioner på hög nivå.
Arbeta med SDK:s resursmodeller i stället för råa svar när det är möjligt.

Undvik att komma åt någon metod som börjar med understreck (_). Enligt konventionen är dessa metoder privata i Python. Det finns inga garantier för saker som brytande ändringar relaterade till offentliga API:er.

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

Hantera sidnumrering korrekt. Iterera alltid över resultat per sida istället för att konvertera till en lista.

# 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

Använd poller.result() för långvariga åtgärder. Använd result() alltid metoden för att säkerställa att åtgärderna har slutförts:

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

Få endast åtkomst till råa svar när det behövs. Använd rå svarsåtkomst sparsamt och endast för specifika krav:

# 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

Feedback

Var den här sidan till hjälp?

Uppdaterades senast 2025-12-09