Memahami jenis respons umum di Azure SDK untuk Python

Azure SDK untuk Python mengabstraksi panggilan ke protokol komunikasi layanan Azure yang mendasari, apakah protokol tersebut adalah HTTP atau AMQP (yang digunakan untuk messaging SDKs seperti ServiceBus dan EventHubs). Misalnya, jika Anda menggunakan salah satu pustaka yang menggunakan HTTP, Azure SDK untuk Python membuat permintaan HTTP dan menerima respons HTTP secara internal. SDK mengabstraksi kompleksitas ini sehingga Anda dapat bekerja dengan objek Python intuitif alih-alih respons HTTP mentah atau payload JSON.

Memahami jenis objek yang Anda terima dari operasi SDK sangat penting untuk menulis aplikasi Azure yang efektif. Artikel ini menjelaskan jenis respons umum yang Anda temui dan bagaimana mereka berhubungan dengan komunikasi HTTP yang mendasar.

Nota

Artikel ini hanya memeriksa skenario HTTP, bukan skenario AMQP.

Objek Python yang dideserialisasi

Azure SDK untuk Python memprioritaskan produktivitas pengembang dengan mengembalikan objek Python yang sangat ditik dari operasi layanan. Alih-alih mengurai JSON atau menangani kode status HTTP secara langsung, Anda bekerja dengan model sumber daya yang mewakili sumber daya Azure sebagai objek Python.

Misalnya, saat Anda mengambil blob dari Azure Storage, Anda menerima BlobProperties objek dengan atribut seperti name, size, dan last_modified bukan kamus JSON mentah:

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

Dari mana data berasal

Memahami aliran data membantu Anda menghargai apa yang dilakukan SDK di belakang layar:

• Kode Anda memanggil metode SDK: Anda memanggil metode seperti get_blob_properties().
• SDK membuat permintaan HTTP: SDK membangun permintaan HTTP yang sesuai dengan parameter header, autentikasi, dan kueri.
• Layanan Azure merespons: Layanan mengembalikan respons HTTP, biasanya dengan payload JSON di isi respons.
• SDK memproses respons: The SDK:
  • Memeriksa kode status HTTP.
  • Mengurai isi respons (biasanya JSON).
  • Memvalidasi data terhadap skema yang diharapkan.
  • Memetakan data ke objek model Python.
• Kode Anda menerima objek Python: Anda bekerja dengan objek yang dideserialisasi, bukan data HTTP mentah.

Abstraksi ini memungkinkan Anda untuk fokus pada logika aplikasi Anda daripada detail protokol HTTP.

Jenis respons umum

Azure SDK untuk Python menggunakan beberapa jenis respons standar di semua layanan. Memahami jenis ini membantu Anda bekerja secara efektif dengan layanan Azure apa pun.

Model sumber daya

Sebagian besar operasi SDK mengembalikan model sumber daya. Objek Python ini mewakili sumber daya Azure. Modelnya khusus layanan tetapi mengikuti pola yang konsisten:

# 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 untuk hasil koleksi

Saat SDK mencantumkan sumber daya, itu mengembalikan objek ItemPaged yang menangani penomoran halaman secara transparan.

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

Mengakses respons HTTP mentah

Meskipun abstraksi tingkat tinggi SDK memenuhi sebagian besar kebutuhan, Anda terkadang memerlukan akses ke respons HTTP yang mendasar. Skenario umum meliputi:

• Memecahkan masalah permintaan yang gagal.
• Mengakses header respons kustom.
• Menerapkan logika ulang khusus.
• Bekerja dengan format respons non-standar.

Sebagian besar metode SDK menerima 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)

Halaman dan iterator

Layanan Azure sering mengembalikan kumpulan sumber daya yang besar. SDK menggunakan ItemPaged untuk menangani koleksi ini secara efisien tanpa memuat semuanya ke dalam memori sekaligus.

Penomoran halaman otomatis

SDK secara otomatis mengambil halaman baru saat Anda melakukan iterasi:

# 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

Bekerja dengan halaman secara spesifik

Anda juga dapat bekerja dengan halaman secara langsung jika diperlukan:

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)

Mengontrol ukuran halaman

Banyak operasi daftar menerima results_per_page parameter:

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

Beberapa metode untuk beberapa layanan Azure memiliki mekanisme lain untuk mengontrol ukuran halaman. Misalnya, Azure Key Vault dan Azure Search menggunakan kwarg top untuk membatasi hasil per panggilan. Untuk contoh yang menggunakan metode Azure Search search() , lihat kode sumber.

Kasus khusus: Operasi dan poller yang berjalan lama

Beberapa operasi Azure tidak dapat segera diselesaikan. Contohnya meliputi:

• Membuat atau menghapus komputer virtual.
• Menyebarkan templat Azure Resource Manager.
• Melatih model pembelajaran mesin
• Menyalin blob berukuran besar.

Operasi ini mengembalikan objek poller yang melacak kemajuan operasi.

Bekerja dengan poller

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

Polling asinkron

Saat Anda menggunakan pola asinkron/menunggu, Anda bekerja dengan 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()

Objek pemantauan untuk contoh operasi berjangka panjang: Mesin virtual

Menyebarkan komputer virtual adalah contoh operasi yang membutuhkan waktu untuk menyelesaikan dan menanganinya dengan mengembalikan objek poller (LROPoller untuk kode sinkron, AsyncLROPoller untuk kode asinkron):

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

Respons akses untuk hasil halaman

Untuk hasil halaman, gunakan by_page() metode dengan 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)

Praktik terbaik

• Lebih suka abstraksi tingkat tinggi.

• Bekerja dengan model sumber daya SDK daripada respons mentah jika memungkinkan.

• Hindari mengakses metode apa pun yang diawali dengan garis bawah (_). Menurut konvensi, metode tersebut bersifat pribadi di Python. Tidak ada jaminan tentang masalah seperti melanggar perubahan dibandingkan dengan API publik:

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

• Tangani pengelolaan halaman dengan benar. Selalu lakukan iterasi atas hasil terpaginasi alih-alih mengonversi menjadi daftar.

  # 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
  

• Gunakan poller.result() untuk operasi yang berjalan lama. Selalu gunakan result() metode untuk memastikan bahwa operasi berhasil diselesaikan:

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

• Akses respons mentah hanya jika diperlukan. Gunakan akses respons mentah dengan hemat dan hanya untuk persyaratan tertentu:

  # 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