tessl/pypi-azure-core

Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Utilities and Helpers

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

Azure Core provides utility functions and helper classes for common operations used across Azure SDK clients. These utilities handle connection string parsing, case-insensitive operations, messaging, and other foundational functionality.

Connection String Utilities

Functions for parsing and handling Azure connection strings commonly used across Azure services.

from azure.core.utils import parse_connection_string
from typing import Dict

def parse_connection_string(conn_str: str) -> Dict[str, str]:
    """
    Parse an Azure connection string into a dictionary.
    
    Parameters:
    - conn_str (str): Connection string in key=value format
    
    Returns:
    Dict[str, str]: Dictionary of connection string components
    """
    ...

Case-Insensitive Collections

Utilities for handling case-insensitive operations commonly needed in HTTP and configuration contexts.

from azure.core.utils import case_insensitive_dict, CaseInsensitiveDict
from typing import Dict, Any, Optional

def case_insensitive_dict(data: Optional[Dict] = None) -> CaseInsensitiveDict:
    """Create a case-insensitive dictionary."""
    ...

class CaseInsensitiveDict(dict):
    """Dictionary that performs case-insensitive key lookups."""
    def __init__(self, data: Optional[Dict] = None): ...
    
    def __getitem__(self, key: str) -> Any: ...
    def __setitem__(self, key: str, value: Any) -> None: ...
    def __delitem__(self, key: str) -> None: ...
    def __contains__(self, key: str) -> bool: ...
    
    def get(self, key: str, default: Any = None) -> Any: ...
    def pop(self, key: str, default: Any = None) -> Any: ...
    def setdefault(self, key: str, default: Any = None) -> Any: ...
    def update(self, other: Dict) -> None: ...

Match Conditions and ETags

Enumeration for conditional request operations commonly used in Azure services.

from azure.core import MatchConditions
from enum import Enum

class MatchConditions(Enum):
    """Conditions for conditional HTTP requests."""
    IfMatch = 1          # If-Match header condition
    IfNoneMatch = 2      # If-None-Match header condition  
    IfModifiedSince = 3  # If-Modified-Since header condition
    IfNotModifiedSince = 4  # If-Not-Modified-Since header condition

Case-Insensitive Enum Support

Metaclass for creating case-insensitive enumerations.

from azure.core import CaseInsensitiveEnumMeta
from enum import EnumMeta

class CaseInsensitiveEnumMeta(EnumMeta):
    """Metaclass for case-insensitive enum lookups."""
    def __getitem__(cls, name: str): ...
    def __call__(cls, value): ...

Serialization Utilities

Utilities for handling serialization, null values, and model operations.

from azure.core.serialization import NULL, AzureJSONEncoder, is_generated_model, as_attribute_dict, attribute_list
from typing import Any, Dict, List
import json

# Null sentinel for JSON serialization
NULL: object  # Falsy sentinel value for null representation

class AzureJSONEncoder(json.JSONEncoder):
    """JSON encoder for Azure-specific types."""
    def default(self, obj: Any) -> Any: ...

def is_generated_model(obj: Any) -> bool:
    """Check if object is a generated model class."""
    ...

def as_attribute_dict(obj: Any) -> Dict[str, Any]:
    """Convert object to attribute dictionary."""
    ...

def attribute_list(obj: Any) -> List[str]:
    """Get list of object attributes."""
    ...

Cloud Events and Messaging

Support for CloudEvents specification and messaging utilities.

from azure.core.messaging import CloudEvent
from typing import Dict, Any, Optional

class CloudEvent:
    """CloudEvents 1.0 schema implementation."""
    def __init__(
        self,
        source: str,
        type: str,
        data: Optional[Any] = None,
        subject: Optional[str] = None,
        **kwargs
    ): ...
    
    @property
    def source(self) -> str: ...
    
    @property
    def type(self) -> str: ...
    
    @property
    def data(self) -> Any: ...
    
    @property
    def subject(self) -> Optional[str]: ...
    
    @property
    def id(self) -> str: ...
    
    @property
    def time(self) -> Optional[str]: ...
    
    @property
    def specversion(self) -> str: ...
    
    @property
    def datacontenttype(self) -> Optional[str]: ...
    
    @property
    def dataschema(self) -> Optional[str]: ...
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert CloudEvent to dictionary."""
        ...
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "CloudEvent":
        """Create CloudEvent from dictionary."""
        ...

Azure Cloud Configuration

Configuration constants for different Azure cloud environments.

from azure.core import AzureClouds

class AzureClouds:
    """Azure cloud environment configurations."""
    
    # Public Azure Cloud
    AZURE_PUBLIC_CLOUD: str
    
    # Azure Government Cloud
    AZURE_GOVERNMENT_CLOUD: str
    
    # Azure China Cloud  
    AZURE_CHINA_CLOUD: str
    
    # Azure German Cloud (deprecated)
    AZURE_GERMAN_CLOUD: str

Global Settings

Access to global Azure SDK configuration settings.

from azure.core.settings import settings, Settings

class Settings:
    """Global Azure SDK settings."""
    
    @property
    def log_level(self) -> str: ...
    
    @log_level.setter  
    def log_level(self, value: str) -> None: ...
    
    @property
    def tracing_enabled(self) -> bool: ...
    
    @tracing_enabled.setter
    def tracing_enabled(self, value: bool) -> None: ...

# Global settings instance
settings: Settings

Usage Examples

Connection String Parsing

from azure.core.utils import parse_connection_string

# Parse Azure Storage connection string
storage_conn_str = "DefaultEndpointsProtocol=https;AccountName=mystorageaccount;AccountKey=mykey;EndpointSuffix=core.windows.net"
parsed = parse_connection_string(storage_conn_str)

print("Account Name:", parsed.get("AccountName"))
print("Account Key:", parsed.get("AccountKey"))
print("Protocol:", parsed.get("DefaultEndpointsProtocol"))
print("Endpoint Suffix:", parsed.get("EndpointSuffix"))

# Parse Service Bus connection string
servicebus_conn_str = "Endpoint=sb://myservicebus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=mykey"
parsed_sb = parse_connection_string(servicebus_conn_str)

print("Endpoint:", parsed_sb.get("Endpoint"))
print("Key Name:", parsed_sb.get("SharedAccessKeyName"))
print("Key:", parsed_sb.get("SharedAccessKey"))

Case-Insensitive Dictionary Operations

from azure.core.utils import CaseInsensitiveDict

# Create case-insensitive dictionary for HTTP headers
headers = CaseInsensitiveDict({
    "Content-Type": "application/json",
    "Authorization": "Bearer token123",
    "User-Agent": "MyApp/1.0"
})

# Access with different casing
print(headers["content-type"])     # Works: "application/json"
print(headers["AUTHORIZATION"])    # Works: "Bearer token123" 
print(headers["user-agent"])       # Works: "MyApp/1.0"

# Update with different casing
headers["CONTENT-TYPE"] = "application/xml"
print(headers["content-type"])     # Updated: "application/xml"

# Check existence with different casing
if "authorization" in headers:
    print("Authorization header present")

# Standard dictionary operations work
headers.update({"Accept": "application/json"})
print(headers.get("ACCEPT", "default"))  # "application/json"

Match Conditions for Conditional Requests

from azure.core import MatchConditions
from azure.core.pipeline.transport import HttpRequest

def create_conditional_request(url, etag, condition):
    """Create HTTP request with conditional headers."""
    request = HttpRequest("GET", url)
    
    if condition == MatchConditions.IfMatch:
        request.headers["If-Match"] = etag
    elif condition == MatchConditions.IfNoneMatch:
        request.headers["If-None-Match"] = etag
    elif condition == MatchConditions.IfModifiedSince:
        request.headers["If-Modified-Since"] = etag  # In this case, etag is a date
    elif condition == MatchConditions.IfNotModifiedSince:
        request.headers["If-Not-Modified-Since"] = etag
    
    return request

# Example usage
etag = '"abc123"'
request = create_conditional_request(
    "https://api.example.com/resource",
    etag,
    MatchConditions.IfNoneMatch
)
print("Headers:", request.headers)

Case-Insensitive Enums

from azure.core import CaseInsensitiveEnumMeta
from enum import Enum

class ApiVersion(Enum, metaclass=CaseInsensitiveEnumMeta):
    """API version enum with case-insensitive lookup."""
    V2020_06_12 = "2020-06-12"
    V2021_04_10 = "2021-04-10"
    V2022_11_02 = "2022-11-02"

# Case-insensitive access
version1 = ApiVersion["v2020_06_12"]      # Works
version2 = ApiVersion["V2020_06_12"]      # Works  
version3 = ApiVersion["2020-06-12"]       # Works

print(f"Version: {version1.value}")  # "2020-06-12"

CloudEvent Handling

from azure.core.messaging import CloudEvent
import json
from datetime import datetime

# Create CloudEvent
event = CloudEvent(
    source="https://myapp.example.com/orders",
    type="com.example.order.created",
    data={"orderId": "12345", "amount": 99.99},
    subject="orders/12345"
)

print(f"Event ID: {event.id}")
print(f"Event Source: {event.source}")
print(f"Event Type: {event.type}")
print(f"Event Data: {event.data}")

# Convert to dictionary for transmission
event_dict = event.to_dict()
print("Event as dict:", json.dumps(event_dict, indent=2))

# Recreate from dictionary
received_event = CloudEvent.from_dict(event_dict)
print(f"Recreated event data: {received_event.data}")

# Handle CloudEvent in webhook receiver
def handle_webhook_event(event_dict):
    """Handle received CloudEvent."""
    try:
        event = CloudEvent.from_dict(event_dict)
        
        if event.type == "com.example.order.created":
            process_order_created(event.data)
        elif event.type == "com.example.order.updated":
            process_order_updated(event.data)
        else:
            print(f"Unknown event type: {event.type}")
            
    except Exception as e:
        print(f"Error processing CloudEvent: {e}")

def process_order_created(order_data):
    print(f"Processing new order: {order_data['orderId']}")

def process_order_updated(order_data):
    print(f"Processing order update: {order_data['orderId']}")

Serialization Utilities

from azure.core.serialization import NULL, AzureJSONEncoder, is_generated_model
import json

# Using NULL sentinel for proper null handling
data = {
    "name": "example",
    "value": NULL,  # Will be serialized as null in JSON
    "active": True
}

# Serialize with Azure JSON encoder
json_string = json.dumps(data, cls=AzureJSONEncoder)
print("Serialized:", json_string)  # {"name": "example", "value": null, "active": true}

# Check if object is generated model
class CustomModel:
    def __init__(self):
        self.name = "test"

model = CustomModel()
if is_generated_model(model):
    print("This is a generated model")
else:
    print("This is a custom model")

Settings Converter Utilities

Utility functions for converting environment variables and configuration values to appropriate types.

from azure.core.settings import convert_bool, convert_logging, convert_azure_cloud
from azure.core import AzureClouds
from typing import Union
import logging

def convert_bool(value: Union[str, bool]) -> bool:
    """Convert string to boolean following Azure SDK conventions.
    
    Args:
        value: The value to convert (bool values returned as-is)
        
    Returns:
        Boolean value matching the intent of the input
    """

def convert_logging(value: Union[str, int]) -> int:
    """Convert string to Python logging level.
    
    Args:
        value: String log level name or integer level
        
    Returns:
        Integer log level for use with logging module
    """

def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:
    """Convert string to AzureClouds enum value.
    
    Args:
        value: String cloud name or AzureClouds enum value
        
    Returns:
        AzureClouds enum value
    """

Usage Examples

from azure.core.settings import convert_bool, convert_logging, convert_azure_cloud
from azure.core import AzureClouds
import os

# Convert environment variables to proper types
debug_mode = convert_bool(os.environ.get("DEBUG", "false"))
print(f"Debug mode: {debug_mode}")  # False

log_level = convert_logging(os.environ.get("LOG_LEVEL", "INFO"))
print(f"Log level: {log_level}")  # 20 (logging.INFO)

cloud = convert_azure_cloud(os.environ.get("AZURE_CLOUD", "AZURE_PUBLIC_CLOUD"))
print(f"Azure cloud: {cloud}")  # AzureClouds.AZURE_PUBLIC_CLOUD

# String conversion examples
print(convert_bool("yes"))     # True
print(convert_bool("1"))       # True
print(convert_bool("on"))      # True
print(convert_bool("true"))    # True
print(convert_bool("false"))   # False

print(convert_logging("DEBUG"))    # 10
print(convert_logging("WARNING"))  # 30
print(convert_logging("ERROR"))    # 40

Enhanced Connection String Parsing

Extended connection string parsing with case sensitivity options.

from azure.core.utils import parse_connection_string
from typing import Mapping

def parse_connection_string(
    conn_str: str, 
    case_sensitive_keys: bool = False
) -> Mapping[str, str]:
    """Parse connection string with optional case preservation.
    
    Args:
        conn_str: String with connection details provided by Azure services
        case_sensitive_keys: Whether to preserve the original casing of keys
        
    Returns:
        Mapping of connection string key/value pairs
    """

Usage Examples

from azure.core.utils import parse_connection_string

# Standard case-insensitive parsing (default)
conn_str = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKey=key123"
result = parse_connection_string(conn_str)
print(result["endpoint"])  # Works - lowercase key

# Case-sensitive parsing preserves original casing
result_sensitive = parse_connection_string(conn_str, case_sensitive_keys=True)
print(result_sensitive["Endpoint"])  # Original casing preserved

Additional CaseInsensitiveDict Methods

Extended functionality for case-insensitive dictionary operations.

from azure.core.utils import CaseInsensitiveDict
from typing import Iterator, Tuple, Any

class CaseInsensitiveDict:
    def copy(self) -> "CaseInsensitiveDict": 
        """Create a shallow copy of the case-insensitive dictionary."""
        
    def lowerkey_items(self) -> Iterator[Tuple[str, Any]]:
        """Iterate over (lowercase_key, value) pairs."""

Usage Examples

from azure.core.utils import CaseInsensitiveDict

headers = CaseInsensitiveDict({
    "Content-Type": "application/json",
    "Authorization": "Bearer token123"
})

# Create a copy
headers_copy = headers.copy()
headers_copy["New-Header"] = "value"

# Iterate with consistent lowercase keys
for key, value in headers.lowerkey_items():
    print(f"{key}: {value}")  # Keys will be lowercase
    # Output:
    # content-type: application/json
    # authorization: Bearer token123

Global Settings Configuration

from azure.core.settings import settings

# Configure global logging level
settings.log_level = "DEBUG"
print(f"Current log level: {settings.log_level}")

# Enable/disable tracing
settings.tracing_enabled = True
print(f"Tracing enabled: {settings.tracing_enabled}")

# Settings are global and affect all Azure SDK clients
def configure_azure_sdk():
    """Configure global Azure SDK settings."""
    settings.log_level = "INFO"
    settings.tracing_enabled = False
    
    # Additional configuration can be added here
    print("Azure SDK configured with custom settings")

configure_azure_sdk()

Cloud Environment Configuration

from azure.core import AzureClouds

def get_cloud_endpoints(cloud_name):
    """Get endpoints for specific Azure cloud."""
    cloud_configs = {
        "public": {
            "base_url": "https://management.azure.com",
            "auth_url": "https://login.microsoftonline.com"
        },
        "government": {
            "base_url": "https://management.usgovcloudapi.net", 
            "auth_url": "https://login.microsoftonline.us"
        },
        "china": {
            "base_url": "https://management.chinacloudapi.cn",
            "auth_url": "https://login.chinacloudapi.cn"
        }
    }
    
    return cloud_configs.get(cloud_name, cloud_configs["public"])

# Usage
public_endpoints = get_cloud_endpoints("public")
gov_endpoints = get_cloud_endpoints("government")

print("Public cloud management URL:", public_endpoints["base_url"])
print("Government cloud auth URL:", gov_endpoints["auth_url"])