Python용 Azure SDK의 일반적인 응답 유형 이해

Python용 Azure SDK는 기본 Azure 서비스 통신 프로토콜에 대한 호출을 추상화합니다. 이 프로토콜은 HTTP 또는 AMQP(예: 및 같은 ServiceBusEventHubs메시징 SDK에 사용됨)입니다. 예를 들어 HTTP를 사용하는 라이브러리 중 하나를 사용하는 경우 Python용 Azure SDK는 HTTP 요청을 수행하고 내부적으로 HTTP 응답을 받습니다. SDK는 이러한 복잡성을 추상화하므로 원시 HTTP 응답 또는 JSON 페이로드 대신 직관적인 Python 개체를 사용할 수 있습니다.

SDK 작업에서 수신하는 개체 유형을 이해하는 것은 효과적인 Azure 애플리케이션을 작성하는 데 필수적입니다. 이 문서에서는 발생하는 일반적인 응답 유형 및 기본 HTTP 통신과 관련된 방법을 설명합니다.

비고

이 문서에서는 AMQP 시나리오가 아닌 HTTP 시나리오만 살펴봅니다.

역직렬화된 Python 개체

Python용 Azure SDK는 서비스 작업에서 강력한 형식의 Python 개체를 반환하여 개발자 생산성을 우선합니다. JSON을 구문 분석하거나 HTTP 상태 코드를 직접 처리하는 대신 Azure 리소스를 Python 개체로 나타내는 리소스 모델을 사용합니다.

예를 들어, Azure Storage에서 Blob을 검색할 때, 원시 JSON 사전이 아닌, BlobProperties, name, size 같은 특성을 가진 last_modified 객체를 받습니다.

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

데이터의 원본 위치

데이터 흐름을 이해하면 SDK가 백그라운드에서 수행하는 작업을 이해할 수 있습니다.

• 코드는 SDK 메서드를 호출합니다. 와 같은 get_blob_properties()메서드를 호출합니다.
• SDK는 HTTP 요청을 생성합니다. SDK는 헤더, 인증 및 쿼리 매개 변수를 사용하여 적절한 HTTP 요청을 빌드합니다.
• Azure 서비스는 다음과 같이 응답합니다. 서비스는 일반적으로 응답 본문에 JSON 페이로드가 있는 HTTP 응답을 반환합니다.
• SDK는 응답을 처리합니다. The SDK:
  • HTTP 상태 코드를 확인합니다.
  • 응답 본문(일반적으로 JSON)을 구문 분석합니다.
  • 예상된 스키마에 대해 데이터의 유효성을 검사합니다.
  • 데이터를 Python 모델 개체에 매핑합니다.
• 코드에서 Python 개체를 받습니다. 원시 HTTP 데이터가 아닌 역직렬화된 개체로 작업합니다.

이 추상화에서는 HTTP 프로토콜 세부 정보 대신 애플리케이션 논리에 집중할 수 있습니다.

일반적인 응답 형식

Python용 Azure SDK는 모든 서비스에서 여러 표준 응답 유형을 사용합니다. 이러한 유형을 이해하면 모든 Azure 서비스에서 효과적으로 작업할 수 있습니다.

리소스 모델

대부분의 SDK 작업은 리소스 모델을 반환합니다. 이러한 Python 개체는 Azure 리소스를 나타냅니다. 모델은 서비스별이지만 일관된 패턴을 따릅니다.

# 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

SDK는 리소스를 나열할 때 페이지 매김을 투명하게 처리하는 개체를 반환 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}")

원시 HTTP 응답에 액세스

SDK의 상위 수준 추상화는 대부분의 요구를 충족하지만 기본 HTTP 응답에 액세스해야 하는 경우도 있습니다. 일반적인 시나리오는 다음과 같습니다.

• 실패한 요청을 디버깅합니다.
• 사용자 지정 응답 헤더에 액세스
• 사용자 지정 재시도 논리 구현
• 비표준 응답 형식으로 작업

대부분의 SDK 메서드는 매개 변수를 허용합니다.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)

페이징 및 반복기

Azure 서비스는 대용량 리소스 컬렉션을 반환하는 경우가 많습니다. SDK는 메모리에 모든 항목을 한 번에 로드하지 않고 이러한 컬렉션을 효율적으로 처리하는 데 사용합니다 ItemPaged .

자동 페이지 매김

SDK는 반복할 때 새 페이지를 자동으로 가져옵니다.

# 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

페이지를 명시적으로 작업하기

필요한 경우 페이지로 직접 작업할 수도 있습니다.

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)

페이지 크기 제어

많은 목록 작업에서 매개 변수를 허용합니다.results_per_page

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

일부 Azure 서비스에 대한 일부 방법에는 페이지 크기를 제어하는 다른 메커니즘이 있습니다. 예를 들어 Azure Key Vault 및 Azure Search는 top kwarg를 사용하여 호출당 결과를 제한합니다. Azure Search search() 메서드를 사용하는 예제는 소스 코드를 참조하세요.

특수 사례: 장기 실행 작업 및 폴러

일부 Azure 작업은 즉시 완료할 수 없습니다. 예를 들면 다음과 같습니다.

• 가상 머신 만들기 또는 삭제
• Azure Resource Manager 템플릿 배포
• 기계 학습 모델 훈련.
• 큰 블롭 복사하기

이러한 작업은 작업의 진행률을 추적하는 폴러 개체를 반환합니다.

폴러와 작업하기

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

비동기 폴러

비동기/await 패턴을 사용할 때 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()

장기 실행 작업의 폴링 객체 예제: 가상 머신

가상 머신 배포는 완료하는 데 시간이 걸리고 폴러 개체(동기 코드, 비동기 코드 LROPoller 의 경우)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")

페이징된 결과에 대한 액세스 응답

페이징된 결과의 경우 다음을 사용하여 메서드by_page()를 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)

모범 사례

• 상위 수준 추상화 선호.

• 가능하면 원시 응답 대신 SDK의 리소스 모델을 사용합니다.

• 밑줄(_)이 접두사로 지정된 메서드에 액세스하지 마세요. 규칙에 따라 이러한 메서드는 Python에서 비공개입니다. 공용 API에 비해 호환성이 손상되는 변경과 같은 문제에 대한 보장은 없습니다.

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

• 페이지 매김을 올바르게 처리합니다. 항상 목록으로 변환하는 대신 페이징된 결과를 반복 처리하세요.

  # 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
  

• 장기 실행 작업에 poller.result()를 사용합니다. 항상 메서드를 result() 사용하여 작업이 성공적으로 완료되도록 합니다.

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

• 필요한 경우에만 원시 응답에 액세스합니다. 특정 요구 사항에 대해서만 원시 응답 액세스를 아끼고 사용합니다.

  # 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