altametris.sara.core.auth.credentials ===================================== .. py:module:: altametris.sara.core.auth.credentials .. autoapi-nested-parse:: 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 ---------- .. autoapisummary:: altametris.sara.core.auth.credentials.logger Classes ------- .. autoapisummary:: altametris.sara.core.auth.credentials.DsCredentials Module Contents --------------- .. py:data:: logger .. py:class:: DsCredentials(config: Optional[altametris.sara.core.auth.config.DsAuthConfig] = 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() .. py:attribute:: config .. py:property:: credential :type: 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 :rtype: TokenCredential :raises DsAuthError: If credential creation fails .. rubric:: Example >>> creds = DsCredentials() >>> token = creds.credential.get_token("https://storage.azure.com/.default") .. py:method:: 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 :rtype: BlobServiceClient :raises DsAuthError: If client creation fails .. rubric:: Example >>> creds = DsCredentials() >>> client = creds.get_blob_service_client() >>> containers = list(client.list_containers()) .. py:method:: get_container_client(container_name: Optional[str] = None) -> azure.storage.blob.ContainerClient Get authenticated ContainerClient for a specific container. :param container_name: Optional container name. If not provided, uses default from config. :returns: Authenticated client for the container :rtype: ContainerClient .. rubric:: Example >>> creds = DsCredentials() >>> # Use default container >>> container = creds.get_container_client() >>> # Use specific container >>> test_container = creds.get_container_client("unit-tests") .. py:method:: get_api_token(scope: Optional[str] = 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. :param scope: Optional OAuth2 scope (uses config default if not provided) Example: "api://ds-dev/.default" :returns: JWT access token :rtype: str :raises DsAuthError: If scope is not configured or token generation fails .. rubric:: 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}"} .. py:method:: validate_access(container_name: Optional[str] = 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 :param container_name: Optional container name to test (uses default if not provided) :returns: True if all validations pass :rtype: bool :raises DsAuthError: If any validation fails .. rubric:: Example >>> creds = DsCredentials() >>> try: ... if creds.validate_access(): ... print("Authentication successful!") ... except DsAuthError as e: ... print(f"Authentication failed: {e}")