altametris.sara.core.auth.credentials

Azure credentials manager for YOLO-SARA.

This module provides a unified credential manager for Azure services, supporting multiple authentication methods and following Azure best practices.

Authentication Flow: 1. Try DefaultAzureCredential (Managed Identity, Service Principal, az login) 2. Fallback to Connection String if configured 3. Raise clear error if all methods fail

Inspired by esv-py-lib patterns for production-ready authentication.

NOTE: This class provides ONLY authentication. Download logic is handled by AzureWeightManager which uses DsCredentials for authentication.

Attributes

logger

Classes

DsCredentials

Unified credential manager for Azure Data Science services.

Module Contents

altametris.sara.core.auth.credentials.logger

class altametris.sara.core.auth.credentials.DsCredentials(config: altametris.sara.core.auth.config.DsAuthConfig | None = None)

Unified credential manager for Azure Data Science services.

This class manages authentication to Azure services (primarily Blob Storage) and provides authenticated clients for Azure SDK operations.

Separation of Concerns: - DsCredentials: Authentication ONLY (tokens, clients) - AzureWeightManager: Download logic + cache integration - CacheManager: Local storage with TTL

Supports three authentication methods (in priority order): 1. DefaultAzureCredential (Managed Identity, Service Principal, Azure CLI) 2. Connection String (local development fallback) 3. Manual credential injection for testing

Example - Basic Usage:

>>> # Automatic configuration from environment
>>> creds = DsCredentials()
>>>
>>> # Get authenticated client (for use with AzureWeightManager)
>>> blob_client = creds.get_blob_service_client()
>>>
>>> # Get API token for APIM authentication
>>> token = creds.get_api_token()

Example - Custom Configuration:

>>> config = DsAuthConfig(
...     storage_account_name="myaccount",
...     weights_container="weights-dev",
...     weights_prefix="my-project"
... )
>>> creds = DsCredentials(config=config)

Example - With Connection String:

>>> import os
>>> os.environ["AZURE_STORAGE_CONNECTION_STRING"] = "DefaultEndpointsProtocol=https;..."
>>> creds = DsCredentials()

config

property credential: azure.core.credentials.TokenCredential

Get Azure credential (lazy loading).

Creates and caches DefaultAzureCredential on first access. DefaultAzureCredential tries multiple authentication methods: 1. Environment variables (Service Principal) 2. Managed Identity 3. Azure CLI (az login) 4. Azure PowerShell 5. Interactive browser

Returns:

Azure credential for authentication

Return type:

TokenCredential

Raises:

DsAuthError – If credential creation fails

Example

>>> creds = DsCredentials()
>>> token = creds.credential.get_token("https://storage.azure.com/.default")

get_blob_service_client() → azure.storage.blob.BlobServiceClient

Get authenticated BlobServiceClient (lazy loading).

Creates and caches BlobServiceClient on first access. Supports both credential-based and connection string authentication.

This client should be used by AzureWeightManager for download operations.

Returns:

Authenticated client for Blob Storage

Return type:

BlobServiceClient

Raises:

DsAuthError – If client creation fails

Example

>>> creds = DsCredentials()
>>> client = creds.get_blob_service_client()
>>> containers = list(client.list_containers())

get_container_client(container_name: str | None = None) → azure.storage.blob.ContainerClient

Get authenticated ContainerClient for a specific container.

Parameters:

container_name – Optional container name. If not provided, uses default from config.

Returns:

Authenticated client for the container

Return type:

ContainerClient

Example

>>> creds = DsCredentials()
>>> # Use default container
>>> container = creds.get_container_client()
>>> # Use specific container
>>> test_container = creds.get_container_client("unit-tests")

get_api_token(scope: str | None = None) → str

Generate an OAuth2 access token for API authentication.

This method generates a JWT token that can be used to authenticate with Azure API Management or other Azure services.

Parameters:

scope – Optional OAuth2 scope (uses config default if not provided) Example: “api://ds-dev/.default”

Returns:

JWT access token

Return type:

str

Raises:

DsAuthError – If scope is not configured or token generation fails

Example

>>> import os
>>> os.environ["DS_API_SCOPE"] = "api://ds-dev/.default"
>>> creds = DsCredentials()
>>> token = creds.get_api_token()
>>> # Use in API requests
>>> headers = {"Authorization": f"Bearer {token}"}

validate_access(container_name: str | None = None) → bool

Validate that authentication and container access work.

This is a convenience method to test the complete authentication chain: 1. Credential creation 2. BlobServiceClient creation 3. Container access

Parameters:

container_name – Optional container name to test (uses default if not provided)

Returns:

True if all validations pass

Return type:

bool

Raises:

DsAuthError – If any validation fails

Example

>>> creds = DsCredentials()
>>> try:
...     if creds.validate_access():
...         print("Authentication successful!")
... except DsAuthError as e:
...     print(f"Authentication failed: {e}")