docs
Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients
npx @tessl/cli install tessl/pypi-azure-core@1.35.0Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients. Azure Core serves as the backbone for all Azure Python SDKs by offering essential components including exception handling, authentication, HTTP pipeline processing, transport abstractions, polling mechanisms, and utilities for building robust Azure service clients.
pip install azure-corefrom azure.core import PipelineClient, AsyncPipelineClient
from azure.core.exceptions import AzureError, HttpResponseError
from azure.core.credentials import TokenCredential, AzureKeyCredentialFor specific components:
from azure.core.pipeline import Pipeline, PipelineRequest, PipelineResponse
from azure.core.pipeline.policies import RetryPolicy, BearerTokenCredentialPolicy
from azure.core.pipeline.transport import HttpRequest, HttpResponsefrom azure.core import PipelineClient
from azure.core.credentials import AzureKeyCredential
from azure.core.pipeline.policies import HeadersPolicy
from azure.core.pipeline.transport import HttpRequest
# Create a pipeline client with authentication
credential = AzureKeyCredential("your-api-key")
client = PipelineClient(
base_url="https://api.example.com",
credential=credential
)
# Create and send a request
request = HttpRequest("GET", "/api/data")
response = client._pipeline.run(request)
# Handle the response
if response.http_response.status_code == 200:
data = response.http_response.json()
print(f"Received data: {data}")
else:
print(f"Request failed with status: {response.http_response.status_code}")Azure Core follows a layered architecture that enables flexibility and extensibility:
This design allows every Azure SDK to customize behavior while maintaining consistency across all Azure services, providing the foundation for the Azure SDK Design Guidelines compliance.
Comprehensive authentication system supporting OAuth tokens, API keys, and SAS credentials with both synchronous and asynchronous interfaces.
class TokenCredential(Protocol):
def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
class AzureKeyCredential:
def __init__(self, key: str): ...
def update(self, key: str) -> None: ...
class AccessToken(NamedTuple):
token: str
expires_on: intAuthentication and Credentials
Modular HTTP processing pipeline with configurable policies for retry logic, authentication, logging, and custom request/response handling.
class Pipeline:
def __init__(self, transport: HttpTransport, policies: List[HTTPPolicy]): ...
def run(self, request: PipelineRequest) -> PipelineResponse: ...
class RetryPolicy(HTTPPolicy):
def __init__(self, retry_total: int = 10, **kwargs): ...
class BearerTokenCredentialPolicy(HTTPPolicy):
def __init__(self, credential: TokenCredential, *scopes: str): ...HTTP transport abstractions supporting multiple async frameworks with configurable connection settings, proxy support, and SSL options.
class HttpTransport(ABC):
def send(self, request: HttpRequest, **kwargs) -> HttpResponse: ...
class HttpRequest:
def __init__(self, method: str, url: str, **kwargs): ...
class HttpResponse:
@property
def status_code(self) -> int: ...
def json(self) -> any: ...High-level REST API building blocks for constructing HTTP requests, handling responses, and integrating with the pipeline system.
from azure.core.rest import HttpRequest, HttpResponse
def send_request(request: HttpRequest, **kwargs) -> HttpResponse: ...Comprehensive exception hierarchy for Azure SDK errors with specialized exceptions for different error conditions and HTTP status codes.
class AzureError(Exception): ...
class HttpResponseError(AzureError): ...
class ResourceNotFoundError(HttpResponseError): ...
class ClientAuthenticationError(HttpResponseError): ...
class ServiceRequestError(AzureError): ...Polling mechanisms for Azure long-running operations with customizable polling strategies and async support.
class LROPoller:
def result(self, timeout: Optional[float] = None) -> Any: ...
def wait(self, timeout: Optional[float] = None) -> None: ...
def done(self) -> bool: ...
class AsyncLROPoller:
async def result(self, timeout: Optional[float] = None) -> Any: ...Polling and Long-Running Operations
Pagination support for handling large result sets with both synchronous and asynchronous iteration patterns.
class ItemPaged:
def by_page(self) -> Iterator[Iterator[T]]: ...
class AsyncItemPaged:
def by_page(self) -> AsyncIterator[AsyncIterator[T]]: ...OpenTelemetry integration for distributed tracing with automatic span creation, custom attributes, and performance monitoring.
class AbstractSpan:
def start(self) -> None: ...
def finish(self) -> None: ...
def add_attribute(self, key: str, value: str) -> None: ...
class TracingOptions:
def __init__(self, **kwargs): ...Distributed Tracing and Diagnostics
Global configuration management for Azure SDK settings with environment variable support and runtime configuration updates.
from azure.core.settings import settings
# Access global settings instance
settings.log_level = "DEBUG"
settings.tracing_enabled = TrueUtility functions and helper classes for common Azure SDK operations including connection string parsing, case-insensitive operations, and message handling.
def parse_connection_string(conn_str: str) -> Dict[str, str]: ...
class CaseInsensitiveDict(dict):
def __init__(self, data: Optional[Dict] = None): ...
class MatchConditions(Enum):
IfMatch = 1
IfNoneMatch = 2
IfModifiedSince = 3
IfNotModifiedSince = 4Comprehensive async/await support with async versions of all core classes, proper resource management, and performance optimization patterns.
class AsyncPipelineClient:
async def send_request(self, request: HttpRequest, **kwargs) -> HttpResponse: ...
class AsyncPipeline:
async def run(self, request: PipelineRequest) -> PipelineResponse: ...