Python için Azure SDK'daki yaygın yanıt türlerini anlama

Python için Azure SDK, temel alınan Azure hizmet iletişim protokolüne yapılan çağrıları özetler; bu protokol HTTP mi yoksa AMQP mi (ve ServiceBusgibi EventHubs mesajlaşma SDK'ları için kullanılır). Örneğin, HTTP kullanan kitaplıklardan birini kullanırsanız, Python için Azure SDK'sı HTTP istekleri yapar ve arka planda HTTP yanıtları alır. SDK, ham HTTP yanıtları veya JSON yükleri yerine sezgisel Python nesneleriyle çalışabilmeniz için bu karmaşıklığı soyutlar.

SDK işlemlerinden aldığınız nesne türlerini anlamak, etkili Azure uygulamaları yazmak için önemlidir. Bu makalede, karşılaştığınız yaygın yanıt türleri ve bunların temel alınan HTTP iletişimi ile ilişkisi açıklanmaktadır.

Seri durumdan çıkarılmış Python nesneleri

Python için Azure SDK, hizmet işlemlerinden kesin olarak yazılan Python nesnelerini döndürerek geliştirici üretkenliğine öncelik verir. JSON ayrıştırma veya HTTP durum kodlarını doğrudan işlemek yerine, Azure kaynaklarını Python nesneleri olarak temsil eden kaynak modelleriyle çalışırsınız.

Örneğin, Azure Depolama'dan bir blob aldığınızda ham JSON BlobProperties sözlüğü yerine , nameve size gibi last_modifiedözniteliklere sahip bir nesne alırsınız:

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

Verilerin geldiği yer

Veri akışını anlamak, SDK'nın arka planda neler yaptığını anlamanıza yardımcı olur:

• Kodunuz bir SDK yöntemini çağırır: gibi get_blob_properties()bir yöntem çağırırsınız.
• SDK bir HTTP isteği oluşturur: SDK, üst bilgiler, kimlik doğrulaması ve sorgu parametreleriyle uygun HTTP isteğini oluşturur.
• Azure hizmeti yanıt verir: Hizmet, genellikle yanıt gövdesinde JSON yükü olan bir HTTP yanıtı döndürür.
• SDK yanıtı işler: The SDK:
  • HTTP durum kodunu denetler.
  • Yanıt gövdesini ayrıştırır (genellikle JSON).
  • Verileri beklenen şemalara göre doğrular.
  • Verileri Python model nesneleriyle eşler.
• Kodunuz Python nesnelerini alır: Ham HTTP verileriyle değil seri durumdan çıkarılmış nesnelerle çalışırsınız.

Bu soyutlama, HTTP protokolü ayrıntıları yerine uygulama mantığınıza odaklanmanızı sağlar.

Yaygın yanıt türleri

Python için Azure SDK tüm hizmetler genelinde birkaç standart yanıt türü kullanır. Bu türleri anlamak, tüm Azure hizmetleriyle etkili bir şekilde çalışmanıza yardımcı olur.

Kaynak modelleri

SDK işlemlerinin çoğu kaynak modellerini döndürür. Bu Python nesneleri Azure kaynaklarını temsil eder. Modeller hizmete özeldir ancak tutarlı desenleri izler:

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

Koleksiyon sonuçları için ItemPaged

SDK kaynakları listelediğinde, sayfalandırmayı saydam bir şekilde işleyen nesneleri döndürür 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}")

Ham HTTP yanıtına erişme

SDK'nın üst düzey soyutlamaları çoğu gereksinimi karşılasa da, bazen temel alınan HTTP yanıtına erişmeniz gerekir. Yaygın senaryolar şunlardır:

• Başarısız isteklerin hata ayıklaması.
• Özel yanıt üst bilgilerine erişme.
• Özel yeniden deneme mantığı uygulama.
• Standart olmayan yanıt biçimleriyle çalışma.

SDK yöntemlerinin çoğu bir raw_response_hook parametreyi kabul eder:

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)

Sayfalama ve yineleyiciler

Azure hizmetleri genellikle büyük kaynak koleksiyonları döndürür. SDK, her şeyi aynı anda belleğe yüklemeden bu koleksiyonları verimli bir şekilde işlemek için kullanır ItemPaged .

Otomatik sayfalandırma

Siz yineledikçe SDK yeni sayfaları otomatik olarak getirir:

# 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

Sayfalarla açık bir şekilde çalışın

Ayrıca gerektiğinde doğrudan sayfalarla da çalışabilirsiniz:

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)

Sayfa boyutunu kontrol et

Birçok liste işlemi bir results_per_page parametreyi kabul etti:

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

Bazı Azure hizmetleri için bazı yöntemlerin sayfa boyutunu denetlemeye yönelik başka mekanizmaları vardır. Örneğin Azure Key Vault ve Azure Search, çağrı başına sonuçları sınırlamak için kwarg kullanır top . Azure Search search() yöntemini kullanan bir örnek için kaynak koduna bakın.

Özel durum: Uzun süre çalışan operasyonlar ve yoklamalar

Bazı Azure işlemleri hemen tamamlanamadı. Örnekler şunları içerir:

• Sanal makineleri oluşturma veya silme.
• Azure Resource Manager şablonlarını dağıtma.
• Makine öğrenmesi modellerini eğitin.
• Büyük veri saklama birimlerini (blob) kopyalama.

Bu işlemler, işlemin ilerleme durumunu izleyen poller nesneleri döndürür.

Anketçilerle çalışma

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

Zaman uyumsuz yoklamalar

async/await desenlerini kullandığınızda, AsyncLROPoller ile çalışırsınız:

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

Uzun süre çalışan işlemler için nesneleri yoklama örneği: Sanal makineler

Sanal makinelerin dağıtılması, tamamlanması zaman alan bir işlem örneğidir ve bu işlem, senkron kod için LROPoller (zaman uyumsuz kod için AsyncLROPoller) poller nesnelerini döndürerek işlenir.

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

Sayfalanmış sonuçlar için erişim tepkisi

Sayfalanmış sonuçlar için ile yöntemini by_page()kullanınraw_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)

En iyi yöntemler

• Üst düzey soyutlamaları tercih edin.

• Mümkün olduğunda ham yanıtlar yerine SDK'nın kaynak modelleriyle çalışın.

• Alt çizgi (_) ön ekli herhangi bir yönteme erişmekten kaçının. Kural gereği, bu yöntemler Python'da özeldir. Genel API'lerle karşılaştırıldığında hataya neden olan değişiklikler gibi sorunlarla ilgili hiçbir garanti yoktur:

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

• Sayfalamayı düzgünce işleyin. Listeye dönüştürmek yerine her zaman sayfalanmış sonuçları yineleyin:

  # 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
  

• Uzun süre çalışan işlemler için poller.result() kullanın. İşlemlerin başarıyla tamamlandığından result() emin olmak için her zaman yöntemini kullanın:

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

• Ham yanıtlara yalnızca gerektiğinde erişin. Ham yanıt erişimini tedbirli ve yalnızca belirli gereksinimler için kullanı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