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

ARM Policies

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

Specialized HTTP pipeline policies for Azure Resource Manager operations. These policies provide ARM-specific functionality including automatic resource provider registration, authentication challenge handling, and enhanced logging with ARM-specific headers.

Capabilities

ARMAutoResourceProviderRegistrationPolicy

Automatically registers unregistered Azure resource providers when encountering 409 errors with MissingSubscriptionRegistration. This policy intercepts failed requests and performs the necessary registration steps transparently.

class ARMAutoResourceProviderRegistrationPolicy(HTTPPolicy[HTTPRequestType, HTTPResponseType]):
    """Auto register an ARM resource provider if not done yet."""
    
    def send(self, request: PipelineRequest[HTTPRequestType]) -> PipelineResponse[HTTPRequestType, HTTPResponseType]: ...

How it Works

Monitors HTTP responses for 409 status codes
Extracts resource provider name from MissingSubscriptionRegistration errors
Automatically registers the resource provider via ARM API
Polls registration status until complete
Retries the original request with a new client request ID

Usage Example

from azure.mgmt.core.policies import ARMAutoResourceProviderRegistrationPolicy
from azure.core.pipeline import Pipeline

# Policy is automatically included in ARMPipelineClient
# Can also be used manually in custom pipelines
policy = ARMAutoResourceProviderRegistrationPolicy()

AsyncARMAutoResourceProviderRegistrationPolicy

Asynchronous version of the resource provider registration policy with the same functionality but using async/await patterns.

class AsyncARMAutoResourceProviderRegistrationPolicy(AsyncHTTPPolicy[HTTPRequestType, AsyncHTTPResponseType]):
    """Auto register an ARM resource provider if not done yet."""
    
    async def send(self, request: PipelineRequest[HTTPRequestType]) -> PipelineResponse[HTTPRequestType, AsyncHTTPResponseType]: ...

ARMChallengeAuthenticationPolicy

Bearer token authentication policy with built-in support for Continuous Access Evaluation (CAE) challenges. Handles ARM-specific authentication patterns and challenge responses.

class ARMChallengeAuthenticationPolicy(BearerTokenCredentialPolicy):
    """Adds a bearer token Authorization header to requests.

    This policy internally handles Continuous Access Evaluation (CAE) challenges. When it can't complete a challenge,
    it will return the 401 (unauthorized) response from ARM.
    """

Usage Example

from azure.mgmt.core.policies import ARMChallengeAuthenticationPolicy
from azure.core.credentials import DefaultAzureCredential

credential = DefaultAzureCredential()
policy = ARMChallengeAuthenticationPolicy(
    credential=credential,
    scopes=["https://management.azure.com/.default"]
)

AsyncARMChallengeAuthenticationPolicy

Asynchronous version of the ARM challenge authentication policy.

class AsyncARMChallengeAuthenticationPolicy(AsyncBearerTokenCredentialPolicy):
    """Adds a bearer token Authorization header to requests.

    This policy internally handles Continuous Access Evaluation (CAE) challenges. When it can't complete a challenge,
    it will return the 401 (unauthorized) response from ARM.
    """

AuxiliaryAuthenticationPolicy

Adds auxiliary authorization tokens to requests via the x-ms-authorization-auxiliary header. Used for scenarios requiring additional authentication tokens alongside the primary bearer token.

class AuxiliaryAuthenticationPolicy(SansIOHTTPPolicy[HTTPRequestType, HTTPResponseType]):
    """Adds auxiliary authorization token header to requests.

    :param auxiliary_credentials: auxiliary credential for authorizing requests
    :type auxiliary_credentials: Sequence[TokenCredential]
    :param str scopes: required authentication scopes
    """
    
    def __init__(self, auxiliary_credentials: Sequence[TokenCredential], *scopes: str, **kwargs: Any) -> None: ...
    def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None: ...

Usage Example

from azure.mgmt.core.policies import AuxiliaryAuthenticationPolicy
from azure.core.credentials import DefaultAzureCredential

# Create auxiliary credentials
aux_credentials = [DefaultAzureCredential()]

policy = AuxiliaryAuthenticationPolicy(
    auxiliary_credentials=aux_credentials,
    "https://management.azure.com/.default"
)

AsyncAuxiliaryAuthenticationPolicy

Asynchronous version of the auxiliary authentication policy.

class AsyncAuxiliaryAuthenticationPolicy(AsyncHTTPPolicy[HTTPRequestType, AsyncHTTPResponseType]):
    """Asynchronous auxiliary authentication policy."""
    
    def __init__(self, auxiliary_credentials: Sequence[AsyncTokenCredential], *scopes: str, **kwargs: Any) -> None: ...
    async def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None: ...
    def on_response(self, request: PipelineRequest[HTTPRequestType], response: PipelineResponse[HTTPRequestType, AsyncHTTPResponseType]) -> Optional[Awaitable[None]]: ...
    def on_exception(self, request: PipelineRequest[HTTPRequestType]) -> None: ...
    async def send(self, request: PipelineRequest[HTTPRequestType]) -> PipelineResponse[HTTPRequestType, AsyncHTTPResponseType]: ...

ARMHttpLoggingPolicy

Enhanced HTTP logging policy that includes ARM-specific headers in the safe headers allowlist. Enables logging of rate limiting headers and other ARM-specific response metadata.

class ARMHttpLoggingPolicy(HttpLoggingPolicy):
    """HttpLoggingPolicy with ARM specific safe headers for loggers."""
    
    DEFAULT_HEADERS_ALLOWLIST = HttpLoggingPolicy.DEFAULT_HEADERS_ALLOWLIST | {
        "x-ms-ratelimit-remaining-subscription-reads",
        "x-ms-ratelimit-remaining-subscription-writes", 
        "x-ms-ratelimit-remaining-tenant-reads",
        "x-ms-ratelimit-remaining-tenant-writes",
        "x-ms-ratelimit-remaining-subscription-resource-requests",
        "x-ms-ratelimit-remaining-subscription-resource-entities-read",
        "x-ms-ratelimit-remaining-tenant-resource-requests",
        "x-ms-ratelimit-remaining-tenant-resource-entities-read",
        "x-ms-ratelimit-remaining-resource",
        "x-ms-request-charge"
    }

Usage Example

from azure.mgmt.core.policies import ARMHttpLoggingPolicy

# Policy is automatically used in ARMPipelineClient
# Can also be configured manually
logging_policy = ARMHttpLoggingPolicy(
    logger=None,  # Uses default logger
    log_request_body=True,
    log_response_body=True
)

Key Features

HTTPS Enforcement

All authentication policies enforce HTTPS for security:

Throws ServiceRequestError for non-HTTPS URLs when enforce_https=True (default)
Can be disabled by setting enforce_https=False in request options

Token Management

Authentication policies handle token lifecycle:

Automatic token refresh when tokens expire (300 second buffer)
Support for multiple auxiliary tokens
Integration with Azure Core credential types

Error Handling

Resource provider registration policy handles and logs registration failures
Authentication policies return appropriate error responses when challenges cannot be completed
All policies maintain error context for debugging

Types

HTTPRequestType = Union[LegacyHttpRequest, HttpRequest]
HTTPResponseType = Union[LegacyHttpResponse, HttpResponse]
AsyncHTTPResponseType = Union[LegacyAsyncHttpResponse, AsyncHttpResponse]
AllHttpResponseType = Union[LegacyHttpResponse, HttpResponse, LegacyAsyncHttpResponse, AsyncHttpResponse]
TokenCredentialType = TypeVar("TokenCredentialType", bound=Union[TokenCredential, AsyncTokenCredential])

Install with Tessl CLI