repositories

azure_bootstrap.repositories

Repository implementations for Azure bootstrap library.

This module contains concrete implementations of configuration and secrets repositories.

Modules:

Name	Description
enhanced_config_repository	Enhanced Config Repository for hierarchical configuration management.
interfaces	Repository interfaces for Azure bootstrap library.
secrets_repository	Secrets Repository implementation for Azure Key Vault integration.

Classes:

Name	Description
EnhancedConfigRepository	Enhanced configuration repository with hierarchical lookup.
SecretsRepository	Implementation of secrets repository for Azure Key Vault.

EnhancedConfigRepository

EnhancedConfigRepository(app_config_connection_string: str | None = None, secrets_repository: SecretsRepositoryInterface | None = None, auto_load_to_environ: bool = False)

Bases: EnhancedConfigRepositoryInterface

Enhanced configuration repository with hierarchical lookup.

This repository provides unified configuration access with automatic precedence: 1. Environment variables (highest priority) 2. Azure App Configuration with automatic Key Vault reference resolution 3. Key Vault secrets (via secrets repository, fallback only) 4. Default values (lowest priority)

Features: - Hierarchical configuration lookup - Azure App Configuration integration with built-in Key Vault resolution - Automatic Key Vault reference resolution when secrets are stored as references - Fallback to direct Key Vault access for non-referenced secrets - Automatic loading to os.environ - Configuration caching for performance - Comprehensive logging

Key Vault Integration: - When secrets are stored in App Config as Key Vault references (JSON format), they are automatically resolved by the Azure SDK - Reference format: {"uri": "https://vault.vault.azure.net/secrets/secretname"} - Requires Managed Identity with "App Configuration Data Reader" and "Key Vault Secrets User" RBAC roles

Usage

With App Config (Key Vault references auto-resolved)

config_repo = EnhancedConfigRepository( app_config_connection_string="...", auto_load_to_environ=True )

Without App Config (environment only)

config_repo = EnhancedConfigRepository() db_host = config_repo.get_value("DATABASE_HOST", "localhost")

Initialize the enhanced configuration repository.

Parameters:

Name	Type	Description	Default
app_config_connection_string	str | None	Azure App Configuration connection string	None
secrets_repository	SecretsRepositoryInterface | None	Optional secrets repository for direct Key Vault fallback	None
auto_load_to_environ	bool	If True, automatically load configs to os.environ on init	False

Methods:

Name	Description
get_value	Get a configuration value with hierarchical lookup.
get_secret_value	Get a secret value from Key Vault.
get_all_values	Get all configuration values from all sources.
load_to_environ	Load configuration values to os.environ.
refresh	Refresh configuration from all sources.
get_repository_metrics	Get metrics about the configuration repository.
clear_cache	Clear the configuration cache.
is_available	Check if the configuration repository is available.
is_app_config_available	Check if Azure App Configuration is available and accessible.
is_key_vault_available	Check if Azure Key Vault is available and accessible.

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def __init__(
    self,
    app_config_connection_string: str | None = None,
    secrets_repository: SecretsRepositoryInterface | None = None,
    auto_load_to_environ: bool = False,
) -> None:
    """
    Initialize the enhanced configuration repository.

    Args:
        app_config_connection_string: Azure App Configuration connection string
        secrets_repository: Optional secrets repository for direct Key Vault fallback
        auto_load_to_environ: If True, automatically load configs to os.environ on init
    """
    self.app_config_connection_string = app_config_connection_string or os.getenv(
        "AZURE_APP_CONFIGURATION_CONNECTION_STRING"
    )
    self.secrets_repository = secrets_repository
    self._cache: dict[str, str] = {}
    self._app_config_available = False
    self._config_provider = None

    # Try to initialize App Configuration provider with Key Vault resolution
    if self.app_config_connection_string:
        try:
            from azure.appconfiguration.provider import load
            from azure.identity import (
                AzureCliCredential,
                ChainedTokenCredential,
                EnvironmentCredential,
                ManagedIdentityCredential,
            )

            # Multi-environment credential chain for Key Vault reference resolution:
            # 1. EnvironmentCredential - Service principal for local dev (AZURE_CLIENT_ID, etc.)
            #    - additionally_allowed_tenants=["*"] allows cross-tenant authentication for Key Vault
            # 2. ManagedIdentityCredential - Azure Functions/App Service with managed identity (production)
            # 3. AzureCliCredential - Fallback for local development with `az login`
            # Order matters: EnvironmentCredential first ensures local dev works immediately
            credential = ChainedTokenCredential(
                EnvironmentCredential(additionally_allowed_tenants=["*"]),
                ManagedIdentityCredential(),
                AzureCliCredential(),
            )

            logger.info(
                "Created ChainedTokenCredential (Environment -> ManagedIdentity -> AzureCli)",
                extra={"operation": "config_init"},
            )

            # Load App Configuration WITH Key Vault credential for resolving Key Vault references
            self._config_provider = load(
                connection_string=self.app_config_connection_string,
                keyvault_credential=credential,
            )
            self._app_config_available = True
            logger.info(
                "Azure App Configuration provider initialized with Key Vault resolution",
                extra={"operation": "config_init"},
            )
        except ImportError as e:
            import sys

            logger.warning(
                f"Azure App Configuration SDK not available (ImportError: {e}), using environment variables only",
                extra={
                    "operation": "config_init",
                    "python_version": sys.version,
                    "python_executable": sys.executable,
                    "import_error": str(e),
                },
            )
        except Exception as e:
            logger.warning(
                f"Failed to initialize App Configuration provider: {type(e).__name__}: {e}, using environment variables",
                extra={
                    "error": str(e),
                    "error_type": type(e).__name__,
                    "operation": "config_init",
                },
                exc_info=True,
            )
    else:
        logger.info(
            "No App Configuration connection string provided, using environment variables only",
            extra={"operation": "config_init"},
        )

    # Auto-load to environment if requested
    if auto_load_to_environ:
        self.load_to_environ()

get_value

get_value(key: str, default: str | None = None) -> str | None

Get a configuration value with hierarchical lookup.

Lookup order: 1. Environment variables (highest priority) 2. Cache (if previously retrieved) 3. Azure App Configuration with automatic Key Vault reference resolution 4. Key Vault secrets (via secrets repository, fallback) 5. Default value (lowest priority)

Note: If a value is stored in App Config as a Key Vault reference, it will be automatically resolved by the provider.

Parameters:

Name	Type	Description	Default
key	str	Configuration key name	required
default	str | None	Default value if key not found	None

Returns:

Type	Description
str | None	Optional[str]: Configuration value or default

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def get_value(self, key: str, default: str | None = None) -> str | None:
    """
    Get a configuration value with hierarchical lookup.

    Lookup order:
    1. Environment variables (highest priority)
    2. Cache (if previously retrieved)
    3. Azure App Configuration with automatic Key Vault reference resolution
    4. Key Vault secrets (via secrets repository, fallback)
    5. Default value (lowest priority)

    Note: If a value is stored in App Config as a Key Vault reference,
    it will be automatically resolved by the provider.

    Args:
        key: Configuration key name
        default: Default value if key not found

    Returns:
        Optional[str]: Configuration value or default
    """
    # 1. Check environment variables first (highest priority)
    env_value = os.getenv(key)
    if env_value is not None:
        logger.debug(f"Config '{key}' found in environment variables")
        return env_value

    # 2. Check cache
    if key in self._cache:
        logger.debug(f"Config '{key}' found in cache")
        return self._cache[key]

    # 3. Try App Configuration provider (with automatic Key Vault resolution)
    if self._config_provider:
        try:
            # The provider acts like a dictionary with automatic Key Vault resolution
            value = self._config_provider.get(key)
            if value is not None:
                # Ensure value is a string (Azure SDK can return Mapping for complex types)
                str_value = str(value) if not isinstance(value, str) else value
                self._cache[key] = str_value
                logger.info(
                    f"Config '{key}' retrieved from App Configuration (Key Vault references auto-resolved)",
                    extra={"key": key, "operation": "get_value"},
                )
                return str_value
        except Exception as e:
            logger.debug(
                f"Config '{key}' not found in App Configuration: {e}",
                extra={"key": key, "error": str(e), "operation": "get_value"},
            )

    # 4. Try secrets repository (fallback for direct Key Vault access)
    if self.secrets_repository:
        secret_value = self.secrets_repository.get_secret(key)
        if secret_value:
            self._cache[key] = secret_value
            logger.info(
                f"Config '{key}' retrieved from direct Key Vault access (fallback)",
                extra={"key": key, "operation": "get_value"},
            )
            return secret_value

    # 5. Return default
    logger.debug(
        f"Config '{key}' not found, using default",
        extra={"key": key, "default": default, "operation": "get_value"},
    )
    return default

get_secret_value

get_secret_value(key: str, default: str | None = None) -> str | None

Get a secret value from Key Vault.

This method specifically targets secrets and bypasses the standard configuration hierarchy to directly access Key Vault.

Parameters:

Name	Type	Description	Default
key	str	Secret key name	required
default	str | None	Default value if secret not found	None

Returns:

Type	Description
str | None	Optional[str]: Secret value or default

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def get_secret_value(self, key: str, default: str | None = None) -> str | None:
    """
    Get a secret value from Key Vault.

    This method specifically targets secrets and bypasses the standard
    configuration hierarchy to directly access Key Vault.

    Args:
        key: Secret key name
        default: Default value if secret not found

    Returns:
        Optional[str]: Secret value or default
    """
    if not self.secrets_repository:
        logger.warning(
            f"No secrets repository available for key '{key}'",
            extra={"key": key, "operation": "get_secret_value"},
        )
        return default

    secret_value = self.secrets_repository.get_secret(key)
    if secret_value:
        logger.info(
            f"Secret '{key}' retrieved successfully",
            extra={"key": key, "operation": "get_secret_value"},
        )
        return secret_value

    logger.debug(
        f"Secret '{key}' not found, using default",
        extra={"key": key, "default": default, "operation": "get_secret_value"},
    )
    return default

get_all_values

get_all_values() -> dict[str, str]

Get all configuration values from all sources.

Returns:

Type	Description
dict[str, str]	Dict[str, str]: All configuration key-value pairs with Key Vault references resolved

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def get_all_values(self) -> dict[str, str]:
    """
    Get all configuration values from all sources.

    Returns:
        Dict[str, str]: All configuration key-value pairs with Key Vault references resolved
    """
    all_configs: dict[str, str] = {}

    # Start with App Configuration provider (Key Vault references already resolved)
    if self._config_provider:
        try:
            # The provider has already loaded and resolved all configs and Key Vault references
            for key in self._config_provider:
                value = self._config_provider.get(key)
                if value is not None:
                    # Ensure value is a string (Azure SDK can return Mapping for complex types)
                    str_value = str(value) if not isinstance(value, str) else value
                    all_configs[key] = str_value
            logger.info(
                f"Retrieved {len(all_configs)} configs from App Configuration (Key Vault refs resolved)",
                extra={"count": len(all_configs), "operation": "get_all_values"},
            )
        except Exception as e:
            logger.error(
                f"Failed to retrieve all configs from App Configuration: {e}",
                extra={"error": str(e), "operation": "get_all_values"},
            )

    # Add cached values
    all_configs.update(self._cache)

    # Environment variables override everything
    for key, value in os.environ.items():
        all_configs[key] = value

    return all_configs

load_to_environ

load_to_environ() -> int

Load configuration values to os.environ.

Precedence Logic: 1. Values already in os.environ are PRESERVED (local.settings.json wins) 2. Missing values are added from App Configuration (fill gaps) 3. This allows local development overrides while providing defaults

Example

local.settings.json sets:

os.environ["USE_MOCK_SHAREPOINT"] = "true"

App Config has:

config["USE_MOCK_SHAREPOINT"] = "false" config["NEW_CONFIG"] = "value"

After load_to_environ():

os.environ["USE_MOCK_SHAREPOINT"] = "true" # ✅ Local preserved os.environ["NEW_CONFIG"] = "value" # ✅ Remote added

Returns:

Name	Type	Description
int	int	Number of NEW values added to os.environ (excludes skipped)

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def load_to_environ(self) -> int:
    """
    Load configuration values to os.environ.

    Precedence Logic:
    1. Values already in os.environ are PRESERVED (local.settings.json wins)
    2. Missing values are added from App Configuration (fill gaps)
    3. This allows local development overrides while providing defaults

    Example:
        # local.settings.json sets:
        os.environ["USE_MOCK_SHAREPOINT"] = "true"

        # App Config has:
        config["USE_MOCK_SHAREPOINT"] = "false"
        config["NEW_CONFIG"] = "value"

        # After load_to_environ():
        os.environ["USE_MOCK_SHAREPOINT"] = "true"  # ✅ Local preserved
        os.environ["NEW_CONFIG"] = "value"  # ✅ Remote added

    Returns:
        int: Number of NEW values added to os.environ (excludes skipped)
    """
    logger.info(
        "Loading all configuration to os.environ",
        extra={"operation": "load_to_environ"},
    )

    added_count = 0
    skipped_count = 0

    # Load from App Configuration provider (Key Vault references already resolved)
    if self._config_provider:
        try:
            for key in self._config_provider:
                value = self._config_provider.get(key)
                if value is None:
                    continue

                # Check if key already exists in os.environ (from local.settings.json)
                if key in os.environ:
                    skipped_count += 1
                    logger.debug(
                        "Skipping key already in os.environ (local override)",
                        extra={
                            "operation": "load_to_environ_skip",
                            "key": key,
                            "local_value": os.environ[key],
                            "remote_value": value,
                        },
                    )
                    continue  # ✅ Preserve local value

                # Key not in os.environ - add from App Config
                # Ensure value is a string (Azure SDK can return Mapping for complex types)
                str_value = str(value) if not isinstance(value, str) else value
                os.environ[key] = str_value
                added_count += 1
                logger.debug(
                    "Added config value to os.environ",
                    extra={
                        "operation": "load_to_environ_add",
                        "key": key,
                    },
                )

            logger.info(
                f"Loaded {added_count} configs from App Configuration to os.environ (Key Vault refs resolved)",
                extra={
                    "count": added_count,
                    "skipped": skipped_count,
                    "operation": "load_to_environ",
                },
            )
        except Exception as e:
            logger.error(
                f"Failed to load configs to environment: {e}",
                extra={"error": str(e), "operation": "load_to_environ"},
            )

    # Load from secrets repository (fallback for direct Key Vault access)
    if self.secrets_repository:
        try:
            secrets = self.secrets_repository.list_secrets()
            for key, value in secrets.items():
                # Check if key already exists in os.environ (from local.settings.json)
                if key in os.environ:
                    skipped_count += 1
                    logger.debug(
                        "Skipping secret already in os.environ (local override)",
                        extra={
                            "operation": "load_to_environ_skip",
                            "key": key,
                            "local_value": os.environ[key],
                        },
                    )
                    continue  # ✅ Preserve local value

                # Key not in os.environ - add from Key Vault
                os.environ[key] = value
                added_count += 1

            logger.info(
                f"Loaded {len(secrets)} secrets from direct Key Vault access (fallback)",
                extra={"count": len(secrets), "operation": "load_to_environ"},
            )
        except Exception as e:
            logger.error(
                f"Failed to load secrets to environment: {e}",
                extra={"error": str(e), "operation": "load_to_environ"},
            )

    logger.info(
        f"Configuration loading complete: {added_count} values added, {skipped_count} local values preserved",
        extra={
            "added_count": added_count,
            "skipped_count": skipped_count,
            "operation": "load_to_environ",
        },
    )

    return added_count

refresh

refresh() -> None

Refresh configuration from all sources.

This method clears the cache and reloads configuration from App Configuration (with Key Vault references resolved) to pick up any changes.

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def refresh(self) -> None:
    """
    Refresh configuration from all sources.

    This method clears the cache and reloads configuration from
    App Configuration (with Key Vault references resolved) to pick up any changes.
    """
    logger.info("Refreshing configuration", extra={"operation": "refresh"})

    # Clear cache
    self._cache.clear()

    # Refresh the App Configuration provider (this will re-fetch and re-resolve Key Vault refs)
    if self._config_provider and hasattr(self._config_provider, "refresh"):
        try:
            self._config_provider.refresh()
            logger.info("App Configuration provider refreshed", extra={"operation": "refresh"})
        except Exception as e:
            logger.error(
                f"Failed to refresh App Configuration provider: {e}",
                extra={"error": str(e), "operation": "refresh"},
            )

    # Clear secrets cache if available
    if self.secrets_repository and hasattr(self.secrets_repository, "clear_cache"):
        self.secrets_repository.clear_cache()

    # Reload to environment
    self.load_to_environ()

    logger.info("Configuration refresh complete", extra={"operation": "refresh"})

get_repository_metrics

get_repository_metrics() -> dict[str, Any]

Get metrics about the configuration repository.

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: Metrics including source counts, cache hits, etc.

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def get_repository_metrics(self) -> dict[str, Any]:
    """
    Get metrics about the configuration repository.

    Returns:
        Dict[str, Any]: Metrics including source counts, cache hits, etc.
    """
    metrics = {
        "app_config_available": self._app_config_available,
        "secrets_repository_available": self.secrets_repository is not None
        and self.secrets_repository.is_available(),
        "cached_keys_count": len(self._cache),
        "environment_variables_count": len(os.environ),
    }

    # Get App Config count if available (includes resolved Key Vault references)
    if self._config_provider:
        try:
            config_count = len(list(self._config_provider))
            metrics["app_config_count"] = config_count
        except Exception:
            metrics["app_config_count"] = 0
    else:
        metrics["app_config_count"] = 0

    # Get secrets count if available (fallback direct access)
    if self.secrets_repository:
        try:
            secrets = self.secrets_repository.list_secrets()
            metrics["secrets_count"] = len(secrets)
        except Exception:
            metrics["secrets_count"] = 0
    else:
        metrics["secrets_count"] = 0

    return metrics

clear_cache

clear_cache() -> None

Clear the configuration cache.

This method clears all cached configuration values, forcing the next get_value call to retrieve fresh data from sources.

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def clear_cache(self) -> None:
    """
    Clear the configuration cache.

    This method clears all cached configuration values, forcing
    the next get_value call to retrieve fresh data from sources.
    """
    logger.info("Clearing configuration cache", extra={"operation": "clear_cache"})
    self._cache.clear()

is_available

is_available() -> bool

Check if the configuration repository is available.

Returns:

Name	Type	Description
bool	bool	True if at least one configuration source is available

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def is_available(self) -> bool:
    """
    Check if the configuration repository is available.

    Returns:
        bool: True if at least one configuration source is available
    """
    return self._app_config_available or (
        self.secrets_repository is not None and self.secrets_repository.is_available()
    )

is_app_config_available

is_app_config_available() -> bool

Check if Azure App Configuration is available and accessible.

Returns:

Name	Type	Description
bool	bool	True if App Config is accessible, False otherwise

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def is_app_config_available(self) -> bool:
    """
    Check if Azure App Configuration is available and accessible.

    Returns:
        bool: True if App Config is accessible, False otherwise
    """
    return self._app_config_available

is_key_vault_available

is_key_vault_available() -> bool

Check if Azure Key Vault is available and accessible.

Returns:

Name	Type	Description
bool	bool	True if Key Vault is accessible, False otherwise

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def is_key_vault_available(self) -> bool:
    """
    Check if Azure Key Vault is available and accessible.

    Returns:
        bool: True if Key Vault is accessible, False otherwise
    """
    if self.secrets_repository:
        return self.secrets_repository.is_available()
    return False

SecretsRepository

SecretsRepository(vault_url: str | None = None)

Bases: SecretsRepositoryInterface

Implementation of secrets repository for Azure Key Vault.

This repository provides access to secrets from Azure Key Vault with graceful fallback to environment variables when Key Vault is not available.

Features: - Azure Key Vault integration (when available) - Environment variable fallback for local development - Caching for performance - Comprehensive logging and error handling

Usage

With Key Vault

secrets_repo = SecretsRepository(vault_url="https://myvault.vault.azure.net/") db_password = secrets_repo.get_secret("database-password")

Without Key Vault (environment variables only)

secrets_repo = SecretsRepository() db_password = secrets_repo.get_secret("DATABASE_PASSWORD")

Initialize the secrets repository.

Parameters:

Name	Type	Description	Default
vault_url	str | None	Optional Azure Key Vault URL (e.g., "https://myvault.vault.azure.net/") If not provided, uses AZURE_KEY_VAULT_URL from environment	None

Methods:

Name	Description
get_secret	Retrieve a secret value by name.
set_secret	Store a secret value (Key Vault only, not environment).
delete_secret	Delete a secret from Key Vault.
list_secrets	List all available secrets (names only for security).
is_available	Check if secrets repository is available and accessible.
clear_cache	Clear the secrets cache.

Source code in azure_bootstrap/repositories/secrets_repository.py

def __init__(self, vault_url: str | None = None) -> None:
    """
    Initialize the secrets repository.

    Args:
        vault_url: Optional Azure Key Vault URL (e.g., "https://myvault.vault.azure.net/")
                  If not provided, uses AZURE_KEY_VAULT_URL from environment
    """
    self.vault_url = vault_url or os.getenv("AZURE_KEY_VAULT_URL")
    self._cache: dict[str, str] = {}
    self._key_vault_available = False
    self._secret_client: Any = None

    # Try to initialize Key Vault client
    if self.vault_url:
        try:
            from azure.identity import DefaultAzureCredential
            from azure.keyvault.secrets import SecretClient

            self._secret_client = SecretClient(
                vault_url=self.vault_url, credential=DefaultAzureCredential()
            )
            self._key_vault_available = True
            logger.info(
                "Azure Key Vault client initialized successfully",
                extra={"vault_url": self.vault_url, "operation": "secrets_init"},
            )
        except ImportError:
            logger.warning(
                "Azure Key Vault SDK not available, using environment variables only",
                extra={"operation": "secrets_init"},
            )
            self._secret_client = None
        except Exception as e:
            logger.warning(
                f"Failed to initialize Key Vault client: {e}, using environment variables",
                extra={"error": str(e), "operation": "secrets_init"},
            )
            self._secret_client = None
    else:
        logger.info(
            "No Key Vault URL provided, using environment variables only",
            extra={"operation": "secrets_init"},
        )
        self._secret_client = None

get_secret

get_secret(secret_name: str) -> str | None

Retrieve a secret value by name.

Lookup order: 1. Cache (if previously retrieved) 2. Azure Key Vault (if available) 3. Environment variables (fallback)

Parameters:

Name	Type	Description	Default
secret_name	str	Name of the secret to retrieve	required

Returns:

Type	Description
str | None	Optional[str]: Secret value if found, None otherwise

Source code in azure_bootstrap/repositories/secrets_repository.py

def get_secret(self, secret_name: str) -> str | None:
    """
    Retrieve a secret value by name.

    Lookup order:
    1. Cache (if previously retrieved)
    2. Azure Key Vault (if available)
    3. Environment variables (fallback)

    Args:
        secret_name: Name of the secret to retrieve

    Returns:
        Optional[str]: Secret value if found, None otherwise
    """
    # Check cache first
    if secret_name in self._cache:
        logger.debug(f"Secret '{secret_name}' found in cache")
        return self._cache[secret_name]

    # Try Key Vault if available
    if self._secret_client:
        try:
            secret = self._secret_client.get_secret(secret_name)
            secret_value: str = secret.value
            self._cache[secret_name] = secret_value
            logger.info(
                f"Secret '{secret_name}' retrieved from Key Vault",
                extra={"secret_name": secret_name, "operation": "get_secret"},
            )
            return secret_value
        except Exception as e:
            logger.warning(
                f"Failed to retrieve secret '{secret_name}' from Key Vault: {e}",
                extra={"secret_name": secret_name, "error": str(e), "operation": "get_secret"},
            )

    # Fallback to environment variables
    # Try both the original name and with underscores replaced by hyphens
    env_value = os.getenv(secret_name) or os.getenv(secret_name.replace("-", "_"))
    if env_value:
        self._cache[secret_name] = env_value
        logger.debug(
            f"Secret '{secret_name}' retrieved from environment variables",
            extra={"secret_name": secret_name, "operation": "get_secret"},
        )
        return env_value

    logger.warning(
        f"Secret '{secret_name}' not found in Key Vault or environment",
        extra={"secret_name": secret_name, "operation": "get_secret"},
    )
    return None

set_secret

set_secret(secret_name: str, secret_value: str) -> bool

Store a secret value (Key Vault only, not environment).

Parameters:

Name	Type	Description	Default
secret_name	str	Name of the secret	required
secret_value	str	Value to store	required

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise

Source code in azure_bootstrap/repositories/secrets_repository.py

def set_secret(self, secret_name: str, secret_value: str) -> bool:
    """
    Store a secret value (Key Vault only, not environment).

    Args:
        secret_name: Name of the secret
        secret_value: Value to store

    Returns:
        bool: True if successful, False otherwise
    """
    if not self._secret_client:
        logger.warning(
            "Cannot set secret without Key Vault client",
            extra={"secret_name": secret_name, "operation": "set_secret"},
        )
        return False

    try:
        self._secret_client.set_secret(secret_name, secret_value)
        self._cache[secret_name] = secret_value  # Update cache
        logger.info(
            f"Secret '{secret_name}' stored in Key Vault",
            extra={"secret_name": secret_name, "operation": "set_secret"},
        )
        return True
    except Exception as e:
        logger.error(
            f"Failed to set secret '{secret_name}': {e}",
            extra={"secret_name": secret_name, "error": str(e), "operation": "set_secret"},
        )
        return False

delete_secret

delete_secret(secret_name: str) -> bool

Delete a secret from Key Vault.

Parameters:

Name	Type	Description	Default
secret_name	str	Name of the secret to delete	required

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise

Source code in azure_bootstrap/repositories/secrets_repository.py

def delete_secret(self, secret_name: str) -> bool:
    """
    Delete a secret from Key Vault.

    Args:
        secret_name: Name of the secret to delete

    Returns:
        bool: True if successful, False otherwise
    """
    if not self._secret_client:
        logger.warning(
            "Cannot delete secret without Key Vault client",
            extra={"secret_name": secret_name, "operation": "delete_secret"},
        )
        return False

    try:
        self._secret_client.begin_delete_secret(secret_name).wait()
        self._cache.pop(secret_name, None)  # Remove from cache
        logger.info(
            f"Secret '{secret_name}' deleted from Key Vault",
            extra={"secret_name": secret_name, "operation": "delete_secret"},
        )
        return True
    except Exception as e:
        logger.error(
            f"Failed to delete secret '{secret_name}': {e}",
            extra={"secret_name": secret_name, "error": str(e), "operation": "delete_secret"},
        )
        return False

list_secrets

list_secrets() -> dict[str, str]

List all available secrets (names only for security).

Returns:

Type	Description
dict[str, str]	Dict[str, str]: Dictionary mapping secret names to metadata (not values)

Source code in azure_bootstrap/repositories/secrets_repository.py

def list_secrets(self) -> dict[str, str]:
    """
    List all available secrets (names only for security).

    Returns:
        Dict[str, str]: Dictionary mapping secret names to metadata (not values)
    """
    secrets_metadata: dict[str, str] = {}

    if self._secret_client:
        try:
            properties = self._secret_client.list_properties_of_secrets()
            for prop in properties:
                if prop.name is not None:
                    secrets_metadata[prop.name] = f"Key Vault (enabled: {prop.enabled})"
            logger.info(
                f"Listed {len(secrets_metadata)} secrets from Key Vault",
                extra={"count": len(secrets_metadata), "operation": "list_secrets"},
            )
        except Exception as e:
            logger.error(
                f"Failed to list secrets from Key Vault: {e}",
                extra={"error": str(e), "operation": "list_secrets"},
            )

    return secrets_metadata

is_available

is_available() -> bool

Check if secrets repository is available and accessible.

Returns:

Name	Type	Description
bool	bool	True if Key Vault client is available, False otherwise

Source code in azure_bootstrap/repositories/secrets_repository.py

def is_available(self) -> bool:
    """
    Check if secrets repository is available and accessible.

    Returns:
        bool: True if Key Vault client is available, False otherwise
    """
    return self._key_vault_available

clear_cache

clear_cache() -> None

Clear the secrets cache.

Source code in azure_bootstrap/repositories/secrets_repository.py

def clear_cache(self) -> None:
    """Clear the secrets cache."""
    self._cache.clear()
    logger.debug("Secrets cache cleared", extra={"operation": "clear_cache"})