tessl/pypi-azure-mgmt-core

Azure management core library defines extensions to Azure Core that are specific to ARM (Azure Resource Management) needed when you use client libraries

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Polling

Name: tessl/pypi-azure-mgmt-core
Author: tessl

ARM-specific polling strategies for managing long-running operations (LROs). These polling implementations handle Azure's asynchronous operation patterns including Azure-AsyncOperation headers, location headers, and body content polling.

Note: The polling classes must be imported directly from their module files rather than from the azure.mgmt.core.polling package due to current package structure limitations.

Capabilities

ARMPolling

Synchronous ARM-specific long-running operation polling that combines multiple polling strategies to handle different ARM LRO patterns.

class ARMPolling(LROBasePolling):
    """ARM-specific synchronous LRO polling."""
    
    def __init__(
        self,
        timeout: float = 30,
        lro_algorithms: Optional[Sequence[LongRunningOperation[HttpRequestTypeVar, AllHttpResponseTypeVar]]] = None,
        lro_options: Optional[Dict[str, Any]] = None,
        path_format_arguments: Optional[Dict[str, str]] = None,
        **operation_config: Any
    ) -> None: ...

Default Polling Algorithms

ARMPolling uses these algorithms in order:

AzureAsyncOperationPolling - Handles Azure-AsyncOperation headers
LocationPolling - Handles Location headers
BodyContentPolling - Polls based on response body content
StatusCheckPolling - Basic status checking fallback

Usage Example

from azure.mgmt.core.polling.arm_polling import ARMPolling
from azure.core.polling import LROPoller

# Create ARM-specific polling
polling = ARMPolling(timeout=60)

# Use with LROPoller
poller = LROPoller(
    client=client,
    initial_response=initial_response,
    deserialization_callback=deserialize_func,
    polling_method=polling
)

# Wait for completion
result = poller.result()

AsyncARMPolling

Asynchronous ARM-specific long-running operation polling with the same polling strategies as ARMPolling but using async/await patterns.

class AsyncARMPolling(AsyncLROBasePolling):
    """ARM-specific asynchronous LRO polling."""
    
    def __init__(
        self,
        timeout: float = 30,
        lro_algorithms: Optional[Sequence[LongRunningOperation[HttpRequestTypeVar, AllHttpResponseTypeVar]]] = None,
        lro_options: Optional[Dict[str, Any]] = None,
        path_format_arguments: Optional[Dict[str, str]] = None,
        **operation_config: Any
    ) -> None: ...

Usage Example

import asyncio
from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling
from azure.core.polling import AsyncLROPoller

async def main():
    # Create async ARM-specific polling
    polling = AsyncARMPolling(timeout=60)
    
    # Use with AsyncLROPoller
    poller = AsyncLROPoller(
        client=client,
        initial_response=initial_response,
        deserialization_callback=deserialize_func,
        polling_method=polling
    )
    
    # Wait for completion
    result = await poller.result()

asyncio.run(main())

AzureAsyncOperationPolling

Specialized polling for operations that return Azure-AsyncOperation headers. Handles Azure's asynchronous operation pattern with support for final-state-via options.

class AzureAsyncOperationPolling(OperationResourcePolling[HttpRequestTypeVar, AllHttpResponseTypeVar]):
    """Implements a operation resource polling, typically from Azure-AsyncOperation."""
    
    def __init__(self, lro_options: Optional[Dict[str, Any]] = None) -> None: ...
    def get_final_get_url(self, pipeline_response: PipelineResponse[HttpRequestTypeVar, AllHttpResponseTypeVar]) -> Optional[str]: ...

LRO Options Support

Supports Swagger LRO options:

final-state-via: azure-async-operation - Uses operation URL for final state
final-state-via: location - Uses location header for final state

Usage Example

from azure.mgmt.core.polling.arm_polling import AzureAsyncOperationPolling

# Configure with LRO options
lro_options = {"final-state-via": "azure-async-operation"}
polling = AzureAsyncOperationPolling(lro_options=lro_options)

BodyContentPolling

Polling strategy that monitors the response body for provisioning state changes. Used for ARM resources that indicate their status through properties.provisioningState.

class BodyContentPolling(LongRunningOperation[HttpRequestTypeVar, AllHttpResponseTypeVar]):
    """Poll based on the body content.

    Implement a ARM resource poller (using provisioning state).
    """
    
    def can_poll(self, pipeline_response: PipelineResponse[HttpRequestTypeVar, AllHttpResponseTypeVar]) -> bool: ...
    def get_polling_url(self) -> str: ...
    def set_initial_status(self, pipeline_response: PipelineResponse[HttpRequestTypeVar, AllHttpResponseTypeVar]) -> str: ...
    def get_status(self, pipeline_response: PipelineResponse[HttpRequestTypeVar, AllHttpResponseTypeVar]) -> str: ...

Supported HTTP Methods

Only polls for requests using:

PUT operations
PATCH operations

Status Code Handling

202: Returns "InProgress"
201: Extracts provisioning state or returns "InProgress"
200: Extracts provisioning state or returns "Succeeded"
204: Returns "Succeeded"

Usage Example

from azure.mgmt.core.polling.arm_polling import BodyContentPolling

# Usually used as part of ARMPolling, but can be used standalone
polling = BodyContentPolling()

# Check if this polling method can handle the response
if polling.can_poll(pipeline_response):
    status = polling.get_status(pipeline_response)

Polling Enums

LRO Options

class _LroOption(str, Enum, metaclass=CaseInsensitiveEnumMeta):
    """Known LRO options from Swagger."""
    FINAL_STATE_VIA = "final-state-via"

Final State Via Options

class _FinalStateViaOption(str, Enum, metaclass=CaseInsensitiveEnumMeta):
    """Possible final-state-via options."""
    AZURE_ASYNC_OPERATION_FINAL_STATE = "azure-async-operation"
    LOCATION_FINAL_STATE = "location"

Key Features

Multi-Strategy Polling

ARM polling tries multiple strategies in sequence:

Check for Azure-AsyncOperation header
Check for Location header
Poll the original URL for body content changes
Fall back to basic status checking

Provisioning State Detection

Body content polling extracts provisioning state from:

{
  "properties": {
    "provisioningState": "Succeeded|Failed|InProgress|..."
  }
}

Timeout and Configuration

All polling methods support:

Custom timeout values (default: 30 seconds)
Custom polling intervals
Path format arguments for URL templating
Additional operation configuration

Error Handling

Throws OperationFailed for invalid initial status codes
Throws BadResponse for empty response bodies when content is expected
Proper error propagation through polling chain

Types

ResponseType = Union[HttpResponse, AsyncHttpResponse]
PipelineResponseType = PipelineResponse[HttpRequest, ResponseType]
HttpRequestType = Union[LegacyHttpRequest, HttpRequest]
AllHttpResponseType = Union[LegacyHttpResponse, HttpResponse, LegacyAsyncHttpResponse, AsyncHttpResponse]
HttpRequestTypeVar = TypeVar("HttpRequestTypeVar", bound=HttpRequestType)
AllHttpResponseTypeVar = TypeVar("AllHttpResponseTypeVar", bound=AllHttpResponseType)

Install with Tessl CLI