A Pythonhoz készült Azure SDK gyakori választípusainak ismertetése

A Pythonhoz készült Azure SDK absztrakciós hívásokat hoz létre az alapul szolgáló Azure-szolgáltatáskommunikációs protokollhoz, függetlenül attól, hogy ez a protokoll HTTP vagy AMQP (amely az SDK-k (például ServiceBus és EventHubs) üzenetküldéséhez használatos). Ha például a HTTP-t használó kódtárak egyikét használja, az Azure SDK for Python HTTP-kéréseket küld, és HTTP-válaszokat fogad a kapucni alatt. Az SDK elvonja ezt a bonyolultságot, így a nyers HTTP-válaszok vagy JSON-hasznos adatok helyett intuitív Python-objektumokkal dolgozhat.

Az SDK-műveletekből kapott objektumtípusok ismerete elengedhetetlen a hatékony Azure-alkalmazások írásához. Ez a cikk ismerteti a gyakran előforduló választípusokat, valamint azt, hogy ezek hogyan kapcsolódnak az alapul szolgáló HTTP-kommunikációhoz.

Megjegyzés:

Ez a cikk csak a HTTP-forgatókönyvet vizsgálja, nem az AMQP-forgatókönyvet.

Deszerializált Python-objektumok

Az Azure SDK for Python úgy rangsorolja a fejlesztői hatékonyságot, hogy erősen gépelt Python-objektumokat ad vissza a szolgáltatásműveletekből. A JSON elemzése vagy a HTTP-állapotkódok közvetlen kezelése helyett olyan erőforrásmodellekkel dolgozik, amelyek Az Azure-erőforrásokat Python-objektumként jelölik.

Ha például lekér egy blobot az Azure Storage-ból, egy BlobProperties olyan attribútumokkal rendelkező objektumot kap, mint a name, size és last_modified nem egy nyers JSON-szótár:

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

Honnan származnak az adatok?

Az adatfolyam megértése segít felmérni, hogy az SDK mit tesz a színfalak mögött:

  • A kód egy SDK-metódust hív meg: Egy olyan metódust hív meg, mint a get_blob_properties().
  • Az SDK egy HTTP-kérést hoz létre: Az SDK létrehozza a megfelelő HTTP-kérést fejlécekkel, hitelesítéssel és lekérdezési paraméterekkel.
  • Az Azure szolgáltatás válaszol: A szolgáltatás EGY HTTP-választ ad vissza, általában egy JSON hasznos adattal a válasz törzsében.
  • Az SDK feldolgozza a választ: Az SDK:
    • Ellenőrzi a HTTP-állapotkódot.
    • Elemzi a válasz törzsét (általában JSON).
    • Ellenőrzi az adatokat a várt sémákkal szemben.
    • Leképezheti az adatokat Python-modellobjektumokra.
  • A kód Python-objektumokat fogad: A deszerializált objektumokkal dolgozik, nem nyers HTTP-adatokkal.

Ez az absztrakció lehetővé teszi, hogy a HTTP protokoll részletei helyett az alkalmazáslogikára összpontosítson.

Gyakori választípusok

A Pythonhoz készült Azure SDK számos szabványos választípust használ az összes szolgáltatásban. Ezeknek a típusoknak a megértése segít hatékonyan dolgozni bármely Azure-szolgáltatással.

Erőforrásmodellek

A legtöbb SDK-művelet erőforrásmodelleket ad vissza. Ezek a Python-objektumok Az Azure-erőforrásokat jelölik. A modellek szolgáltatásspecifikusak, de konzisztens mintákat követnek:

# 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 a gyűjtemény eredményeihez

Amikor az SDK felsorolja az erőforrásokat, a lapozást transzparensen kezelő objektumokat ad vissza 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}")

A nyers HTTP-válasz elérése

Bár az SDK magas szintű absztrakciói megfelelnek a legtöbb igénynek, néha hozzá kell férnie a mögöttes HTTP-válaszhoz. Gyakori forgatókönyvek a következők:

  • Sikertelen kérések hibakeresése.
  • Egyéni válaszfejlécek elérése.
  • Egyéni újrapróbálkozási logika implementálása.
  • Nem szabványos válaszformátumok használata.

A legtöbb SDK-metódus elfogad egy paramétert 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)

Lapozás és iterátorok

Az Azure-szolgáltatások gyakran nagy mennyiségű erőforrást adnak vissza. Az SDK ItemPaged hatékonyan kezeli ezeket a gyűjteményeket anélkül, hogy mindent egyszerre tölt be a memóriába.

Automatikus lapozás

Az SDK automatikusan beolvassa az új lapokat az iteráció során:

# 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

Lapokkal explicit módon dolgozzon

Szükség esetén közvetlenül is használhatja a lapokat:

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)

Oldalméret szabályozása

Számos listaművelet fogad el egy paramétert results_per_page :

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

Egyes Azure-szolgáltatások egyes metódusai más mechanizmusokkal is rendelkeznek az oldalméret szabályozására. Az Azure Key Vault és az Azure Search például a kwarg használatával korlátozza a top hívásonkénti eredményeket. Az Azure Search search() metódust használó példáért tekintse meg a forráskódot.

Speciális eset: Hosszan futó műveletek és lekérdezések

Egyes Azure-műveletek nem fejezhetők be azonnal. Ide sorolhatóak például a következők:

  • Virtuális gépek létrehozása vagy törlése.
  • Azure Resource Manager-sablonok üzembe helyezése.
  • Gépi tanulási modellek betanítása.
  • Nagy adatblobok másolása.

Ezek a műveletek olyan lekérdezési objektumokat ad vissza, amelyek nyomon követik a művelet előrehaladását.

Dolgozzon együtt lekérdezőkkel

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

Aszinkron pollerek

Ha aszinkron/várakozási mintákat használ, a következőkkel AsyncLROPollerdolgozik:

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

Objektumok lekérdezése hosszú ideig futó műveletekhez– példa: Virtuális gépek

A virtuális gépek üzembe helyezése olyan művelet, amely időt vesz igénybe, és pollerobjektumok visszaadásával kezeli azt szinkron kód esetén (LROPoller), illetve aszinkron kód esetén (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")

Hozzáférési válasz lapozott eredményekhez

Lapozott eredmények esetén használja a by_page() következő metódust 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)

Ajánlott eljárások

  • Előnyben részesítse a magas szintű absztrakciókat.

  • Amikor csak lehetséges, dolgozzon az SDK erőforrásmodelljeivel a nyers válaszok helyett.

  • Ne érje el az aláhúzásjellel (_) előtaggal ellátott metódusokat. Ezek a metódusok konvenció szerint privátak a Pythonban. Az olyan problémák esetén, mint a nyilvános API-khoz képest történő kompatibilitásbontó változások, nincsenek garanciák.

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

  • A lapozás megfelelő kezelése. Mindig iterálja az oldalakra bontott eredményeket ahelyett, hogy listává alakítaná azokat.

    # 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

  • Használja a poller.result() függvényt a hosszan futó műveletekhez. Mindig használja a result() metódust, hogy a műveletek sikeresen befejeződjenek:

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

  • A nyers válaszokat csak szükség esetén érheti el. A nyers válaszhozzáférés használata takarékosan és csak meghatározott követelmények esetén:

    # 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
  • Utolsó frissítés időpontja: