azure_bootstrap

azure_bootstrap

Azure Bootstrap Library

A production-ready Azure bootstrap library that handles application initialization for Azure Functions, including App Configuration, Key Vault, and App Insights integration.

This library solves the circular dependency between logging and configuration by: 1. Starting with basic console logging 2. Loading configuration from Azure App Configuration + Key Vault 3. Upgrading to App Insights telemetry when available 4. Loading all configs to os.environ for transparent access

Quick Start

from azure_bootstrap import initialize_application, get_bootstrap_logger

Get logger that works immediately

logger = get_bootstrap_logger(name)

Bootstrap the application (App Config + Key Vault + App Insights)

config_repo = initialize_application()

Now all configs are in os.environ

db_host = os.getenv("DATABASE_HOST")

For detailed usage, see: https://github.com/TheViziusGroup/azure-bootstrap

Modules:

Name	Description
aks	AKS runtime helpers — graceful shutdown, build info, probes, pod context.
alerts	Tier 2 tiered-alert dispatcher.
audit	Audit log conventions.
auth	Webhook + API-key authentication helpers.
bootstrap	v1-compatible wrappers and helpers.
config_refresh	Dynamic log-level refresh via App Configuration.
contrib	Package-data templates for the scaffold CLI.
counters	Best-effort process-local counters.
db	Relational database access layer — SQLAlchemy 2 engine/session factory.
documentdb	MongoDB / Cosmos DB document database access.
email	Azure Communication Services email sender.
exceptions	Project-neutral exception hierarchy for pipelined applications.
failclose	Fail-closed-for-auth / fail-open-for-features env helpers.
fastapi_middleware	FastAPI request-timing + alerting middleware.
governance	Governance — budget guards and usage tracking.
health	Framework-neutral readiness probes for Azure App Configuration + App Insights.
heartbeat	Background heartbeat + worker-progress watchdog.
http	Hardened outbound HTTP client — sync requests stack.
identity	Workload Identity / DefaultAzureCredential wrapper.
ingress	Tier 2 attachment / file-upload hardening.
logging	Tier 1 logging primitives.
metrics	Aggregate library metrics for /api/metrics endpoints.
models	Custom exceptions for bootstrap operations.
notify	Sender-facing throttling + two-tier notification builders.
openai	SDK-agnostic Azure OpenAI / Anthropic usage tracker.
path_safety	Filename + path sanitization.
pdf_safety	PDF action stripping for untrusted pass-through.
phases	Per-phase guarded execution for multi-stage pipelines.
ratelimit	In-process token-bucket rate limiter.
repositories	Repository implementations for Azure bootstrap library.
retry	Tenacity-backed retry wrappers with built-in counter bumps and logging.
sb_lock	Service Bus message-lock management.
scheduler	NCRONTAB → APScheduler CronTrigger parser.
security	Constant-time comparison + FastAPI API-key helper.
servicebus	Tier 3 Service Bus helpers: consumer watchdog, DLQ growth alarm, daily digest.
services	Service implementations for Azure bootstrap library.
softfail	Helpers for the "log + alert + continue with degraded result" pattern.
subscription	Generic external-resource renewal loop pattern.
tokens	HMAC-SHA256 action-token signer (generalized DLQ-resubmit pattern).
tracing	Tier 1 tracing primitives: @traced decorator, latency histograms, slow thresholds.
transports	Logging transport registry — choose where logs go, toggle each independently.
validation	Lightweight JSON / queue-message schema validation.

Classes:

Name	Description
AcsEmailSender	Transactional email sender via Azure Communication Services.
InvalidMessageError	Queue payload failed schema validation.
NetworkError	Connection / timeout failure against an external service.
PipelineError	Base for any application-level pipeline failure.
RateLimitError	HTTP 429 or equivalent — caller should back off.
TransientError	Marker for known-recoverable failures.
UnrecoverableError	Marker: this failure will never succeed on retry.
JsonLogFormatter	Render a log record as a single line of JSON.
ConfigurationError	Exception raised when configuration loading or access fails.
KeyVaultError	Exception raised when Key Vault operations fail.
RepositoryError	Base exception for repository errors.
EnhancedConfigRepository	Enhanced configuration repository with hierarchical lookup.
EnhancedConfigRepositoryInterface	Abstract interface for enhanced configuration repository operations.
SecretsRepositoryInterface	Abstract interface for secrets repository operations.
SecretsRepository	Implementation of secrets repository for Azure Key Vault.
ApplicationBootstrap	Orchestrates the complete application startup sequence with proper logging flow.
BootstrapLogger	Bootstrap logging manager that provides safe logging before full configuration.
ExtraFieldsFormatter	Custom formatter that appends extra fields from log records to the message.
ApplicationBootstrapInterface	Interface for application bootstrap orchestrator.
BootstrapLoggerInterface	Interface for bootstrap logging manager.
TelemetryManagerInterface	Interface for telemetry manager.
TelemetryManager	Manages Application Insights telemetry and structured logging

Functions:

Name	Description
build_info	Return version/build metadata from downward-API or CI env vars.
verify_chain	Verify the integrity of an ordered list of :class:ChainedAuditRecord objects.
verify_hmac_signature	Constant-time HMAC-SHA256 verify (GitHub/Sumo sha256=… style).
bootstrap_initialized	Process-local state probe — useful for /api/health/ready endpoints.
ensure_bootstrap	Lazy idempotent wrapper around v1 initialize_application.
load_local_settings	Load env vars from an Azure-Functions-style local.settings.json.
bump_counter	Thread-safe increment. Never raises.
counter_snapshot	Return a copy of the counter map.
drain_outbox	Claim and drain pending outbox rows. Returns count sent.
is_unrecoverable	The consumer's classifier — single isinstance check.
build_session	Return a requests.Session with urllib3 Retry mounted.
request_with_retry	Issue an HTTP request with default timeout, traceparent, and SSRF guard.
build_tenant_credential	Build a WorkloadIdentityCredential scoped to tenant_id.
build_tenant_credential_cached	Return an access token for tenant_id / scope, using the cache.
configure_logging	Install structured-logging defaults. Idempotent — replaces handlers.
correlation_scope	Push correlation context for the duration of the with-block.
mask_secrets_in_dict	Shallow copy with secret-keyed values replaced by '***'.
sanitize_for_log	Replace control chars (\x00-\x1f, \x7f) with '?' and truncate.
confine_to_root	Resolve raw and assert it's under allowed_root.
sanitize_path_segment	Normalize a single filename / path segment.
run_phase	Run fn under a try/except. NEVER re-raises.
run_phases	Run a list of (name, callable) pairs in order, swallowing per-phase
create_enhanced_config_repository	Factory function to create an enhanced configuration repository.
compare_secrets	Constant-time equality. Returns False on any None / empty input.
initialize_application	Convenience function to perform complete application bootstrap.
ensure_bootstrap_logging	Ensure bootstrap logging is configured.
get_bootstrap_logger	Get a logger that works during bootstrap phase.
soft_fail	Context-manager form of :func:soft_fail_with.
soft_fail_with	Call fn(*args, **kwargs); soft-fail with fallback on caught error.
traced	Trace sync or async functions.
configure_transports	Enable/disable the ten built-in transports. Idempotent and re-runnable.
disable_transport	Detach (and close) the transport's handler. Idempotent.
enable_transport	Attach the transport's handler to the root logger. Idempotent.
list_transports	Return {name: {"registered": True, "enabled": bool}} for all transports.
register_transport	Register a transport factory under name.
queue_message_schema	Build a sensible MessageSchema for the common consumer case.
validate_message	Validate payload against schema. Returns the dict on success.
refresh_setting	Re-read named settings from the cached App Configuration repo and

AcsEmailSender

AcsEmailSender(*, connection_string: str | None = None, sender_address: str | None = None)

Transactional email sender via Azure Communication Services.

Methods:

Name	Description
send	Send an email; returns message id on success.

Source code in azure_bootstrap/email/__init__.py

def __init__(
    self,
    *,
    connection_string: str | None = None,
    sender_address: str | None = None,
) -> None:
    self._connection_string = connection_string or require_env("ACS_CONNECTION_STRING")
    self._sender = sender_address or require_env("ACS_SENDER_ADDRESS")
    self._client: Any = None

send

send(*, to: list[str], subject: str, html_body: str, plain_text: str | None = None) -> str

Send an email; returns message id on success.

Source code in azure_bootstrap/email/__init__.py

def send(
    self,
    *,
    to: list[str],
    subject: str,
    html_body: str,
    plain_text: str | None = None,
) -> str:
    """Send an email; returns message id on success."""
    message: dict[str, Any] = {
        "senderAddress": self._sender,
        "recipients": {"to": [{"address": addr} for addr in to]},
        "content": {"subject": subject, "html": html_body},
    }
    if plain_text:
        message["content"]["plainText"] = plain_text
    poller = (
        self._client.begin_send(message)
        if self._client
        else self._get_client().begin_send(message)
    )
    result = poller.result()
    msg_id = getattr(result, "id", None) or getattr(result, "message_id", "unknown")
    _logger.info("ACS email sent id=%s to=%s", msg_id, to)
    return str(msg_id)

InvalidMessageError

Bases: UnrecoverableError

Queue payload failed schema validation.

NetworkError

Bases: TransientError

Connection / timeout failure against an external service.

PipelineError

Bases: Exception

Base for any application-level pipeline failure.

RateLimitError

Bases: TransientError

HTTP 429 or equivalent — caller should back off.

TransientError

Bases: PipelineError

Marker for known-recoverable failures.

Not in UnrecoverableError's tree, so consumers automatically retry.

UnrecoverableError

Bases: PipelineError

Marker: this failure will never succeed on retry.

Consumers MUST dead-letter messages whose handlers raised any subclass of UnrecoverableError. Anything else is transient (abandon → broker redelivers).

JsonLogFormatter

JsonLogFormatter(*, ensure_ascii: bool = False, mask_extras: bool = True)

Bases: Formatter

Render a log record as a single line of JSON.

Emitted fields: timestamp (ISO-8601 UTC), level, logger, message, exception (only when exc_info is present), plus every non-reserved extra={} field (correlation IDs included, since :class:CorrelationFilter attaches them as record attributes). Extra fields are passed through :func:mask_secrets_in_dict so secret-keyed values are redacted to ***.

Source code in azure_bootstrap/logging/jsonformatter.py

def __init__(self, *, ensure_ascii: bool = False, mask_extras: bool = True) -> None:
    super().__init__()
    self._ensure_ascii = ensure_ascii
    self._mask_extras = mask_extras

ConfigurationError

Bases: RepositoryError

Exception raised when configuration loading or access fails.

This exception is raised when: - Azure App Configuration connection fails - Configuration values are missing or invalid - Configuration refresh operations fail - Configuration loading encounters critical errors

KeyVaultError

Bases: RepositoryError

Exception raised when Key Vault operations fail.

This exception is raised when: - Key Vault connection or authentication fails - Secret retrieval operations fail - Key Vault access is denied or unavailable

RepositoryError

Bases: Exception

Base exception for repository errors.

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

EnhancedConfigRepositoryInterface

Bases: ABC

Abstract interface for enhanced configuration repository operations.

This interface defines all configuration access operations for integration with Azure App Configuration, Key Vault, and environment variables following a hierarchical precedence model.

Configuration Precedence (highest to lowest): 1. Environment variables (os.environ) 2. Azure App Configuration 3. Key Vault secrets (via secrets repository) 4. Default values

Benefits: - Abstracts configuration storage implementation details - Enables dependency injection and testing with mocks - Provides clear contract for configuration access - Supports multiple backends with transparent fallback

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 all configuration values into os.environ.
refresh	Refresh configuration from all sources.
get_repository_metrics	Get metrics about the configuration repository.
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.

get_value abstractmethod

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

Get a configuration value with hierarchical lookup.

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

Raises:

Type	Description
ConfigurationError	If configuration access fails

Source code in azure_bootstrap/repositories/interfaces/enhanced_config_repository_interface.py

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

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

    Returns:
        Optional[str]: Configuration value or default

    Raises:
        ConfigurationError: If configuration access fails
    """
    pass

get_secret_value abstractmethod

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

Raises:

Type	Description
SecretAccessError	If secret retrieval fails

Source code in azure_bootstrap/repositories/interfaces/enhanced_config_repository_interface.py

@abstractmethod
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

    Raises:
        SecretAccessError: If secret retrieval fails
    """
    pass

get_all_values abstractmethod

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

Raises:

Type	Description
ConfigurationError	If configuration access fails

Source code in azure_bootstrap/repositories/interfaces/enhanced_config_repository_interface.py

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

    Returns:
        Dict[str, str]: All configuration key-value pairs

    Raises:
        ConfigurationError: If configuration access fails
    """
    pass

load_to_environ abstractmethod

load_to_environ() -> int

Load all configuration values into os.environ.

This method loads all configuration from App Config and Key Vault into environment variables for transparent application access.

Returns:

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

Raises:

Type	Description
ConfigurationError	If loading fails

Source code in azure_bootstrap/repositories/interfaces/enhanced_config_repository_interface.py

@abstractmethod
def load_to_environ(self) -> int:
    """
    Load all configuration values into os.environ.

    This method loads all configuration from App Config and Key Vault
    into environment variables for transparent application access.

    Returns:
        int: Number of new values added to os.environ (excludes skipped values)

    Raises:
        ConfigurationError: If loading fails
    """
    pass

refresh abstractmethod

refresh() -> None

Refresh configuration from all sources.

This method reloads configuration from App Configuration and Key Vault to pick up any changes.

Raises:

Type	Description
ConfigurationError	If refresh fails

Source code in azure_bootstrap/repositories/interfaces/enhanced_config_repository_interface.py

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

    This method reloads configuration from App Configuration and
    Key Vault to pick up any changes.

    Raises:
        ConfigurationError: If refresh fails
    """
    pass

get_repository_metrics abstractmethod

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/interfaces/enhanced_config_repository_interface.py

@abstractmethod
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.
    """
    pass

is_app_config_available abstractmethod

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/interfaces/enhanced_config_repository_interface.py

@abstractmethod
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
    """
    pass

is_key_vault_available abstractmethod

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/interfaces/enhanced_config_repository_interface.py

@abstractmethod
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
    """
    pass

SecretsRepositoryInterface

Bases: ABC

Abstract interface for secrets repository operations.

This interface defines all secret access operations for integration with Azure Key Vault or other secret storage backends.

Benefits: - Abstracts secret storage implementation details from business logic - Enables dependency injection and testing with mocks - Provides clear contract for secret access operations - Supports multiple backends (Key Vault, local files, environment, etc.)

Methods:

Name	Description
get_secret	Retrieve a secret value by name.
set_secret	Store a secret value (if backend supports writes).
delete_secret	Delete a secret (if backend supports deletion).
list_secrets	List all available secrets (names only for security).
is_available	Check if secrets repository is available and accessible.

get_secret abstractmethod

get_secret(secret_name: str) -> str | None

Retrieve a secret value by name.

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

Raises:

Type	Description
SecretAccessError	If secret retrieval fails

Source code in azure_bootstrap/repositories/interfaces/secrets_repository_interface.py

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

    Args:
        secret_name: Name of the secret to retrieve

    Returns:
        Optional[str]: Secret value if found, None otherwise

    Raises:
        SecretAccessError: If secret retrieval fails
    """
    pass

set_secret abstractmethod

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

Store a secret value (if backend supports writes).

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

Raises:

Type	Description
SecretAccessError	If secret storage fails
NotImplementedError	If backend doesn't support writes

Source code in azure_bootstrap/repositories/interfaces/secrets_repository_interface.py

@abstractmethod
def set_secret(self, secret_name: str, secret_value: str) -> bool:
    """
    Store a secret value (if backend supports writes).

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

    Returns:
        bool: True if successful, False otherwise

    Raises:
        SecretAccessError: If secret storage fails
        NotImplementedError: If backend doesn't support writes
    """
    pass

delete_secret abstractmethod

delete_secret(secret_name: str) -> bool

Delete a secret (if backend supports deletion).

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

Raises:

Type	Description
SecretAccessError	If secret deletion fails
NotImplementedError	If backend doesn't support deletion

Source code in azure_bootstrap/repositories/interfaces/secrets_repository_interface.py

@abstractmethod
def delete_secret(self, secret_name: str) -> bool:
    """
    Delete a secret (if backend supports deletion).

    Args:
        secret_name: Name of the secret to delete

    Returns:
        bool: True if successful, False otherwise

    Raises:
        SecretAccessError: If secret deletion fails
        NotImplementedError: If backend doesn't support deletion
    """
    pass

list_secrets abstractmethod

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

Raises:

Type	Description
SecretAccessError	If listing fails

Source code in azure_bootstrap/repositories/interfaces/secrets_repository_interface.py

@abstractmethod
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

    Raises:
        SecretAccessError: If listing fails
    """
    pass

is_available abstractmethod

is_available() -> bool

Check if secrets repository is available and accessible.

Returns:

Name	Type	Description
bool	bool	True if repository is accessible, False otherwise

Source code in azure_bootstrap/repositories/interfaces/secrets_repository_interface.py

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

    Returns:
        bool: True if repository is accessible, False otherwise
    """
    pass

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"})

ApplicationBootstrap

ApplicationBootstrap(secrets_repository: SecretsRepositoryInterface | None = None)

Bases: ApplicationBootstrapInterface

Orchestrates the complete application startup sequence with proper logging flow.

This class handles the complex bootstrap process that requires careful ordering to avoid circular dependencies between logging, configuration, and secrets.

Bootstrap Flow

1. Initialize console logging (safe fallback that always works)
2. Try App Insights from environment variables if available
3. Create and load enhanced configuration from App Config/Key Vault
4. Attempt to upgrade logging to App Insights if connection string is now available
5. Load all configuration to os.environ for transparent application access
6. Log completion and provide access to configured components

Key Features

• Eliminates circular dependencies between logging and configuration
• Provides working logging throughout the entire bootstrap process
• Graceful fallbacks at every step ensure robustness
• Comprehensive logging of bootstrap progress and decisions
• Transparent configuration access via os.environ after completion

Usage Example

Simple bootstrap

bootstrap = ApplicationBootstrap() config_repo = bootstrap.initialize()

Bootstrap with secrets repository

bootstrap = ApplicationBootstrap(secrets_repository=my_secrets_repo) config_repo = bootstrap.initialize()

After bootstrap, all configs are in os.environ

app_insights_key = os.environ.get("APPLICATIONINSIGHTS_CONNECTION_STRING") database_host = os.getenv("DATABASE_HOST")

Initialize the application bootstrap orchestrator.

Parameters:

Name	Type	Description	Default
secrets_repository	SecretsRepositoryInterface | None	Optional secrets repository for Key Vault integration	None

Methods:

Name	Description
initialize	Execute the complete bootstrap sequence with proper logging flow.
get_config_repository	Get the configured repository after bootstrap.
is_bootstrap_completed	Check if bootstrap process has completed successfully.

Source code in azure_bootstrap/services/application_bootstrap.py

def __init__(self, secrets_repository: SecretsRepositoryInterface | None = None) -> None:
    """
    Initialize the application bootstrap orchestrator.

    Args:
        secrets_repository: Optional secrets repository for Key Vault integration
    """
    # Start with bootstrap logging immediately
    BootstrapLogger.configure_bootstrap_logging()
    self.logger = get_bootstrap_logger(__name__)

    self.secrets_repository = secrets_repository
    self.config_repository: EnhancedConfigRepositoryInterface | None = None
    self._bootstrap_completed = False

    self.logger.info(
        "ApplicationBootstrap initialized",
        extra={
            "has_secrets_repository": bool(secrets_repository),
            "operation": "bootstrap_init",
            "component": "ApplicationBootstrap",
        },
    )

initialize

initialize() -> EnhancedConfigRepositoryInterface

Execute the complete bootstrap sequence with proper logging flow.

This method orchestrates the entire application startup process following the correct sequence to avoid circular dependencies while ensuring working logging throughout.

Returns:

Type	Description
EnhancedConfigRepositoryInterface	Configured enhanced configuration repository with all configs loaded

Raises:

Type	Description
RuntimeError	If bootstrap fails in an unrecoverable way

Source code in azure_bootstrap/services/application_bootstrap.py

def initialize(self) -> EnhancedConfigRepositoryInterface:
    """
    Execute the complete bootstrap sequence with proper logging flow.

    This method orchestrates the entire application startup process following
    the correct sequence to avoid circular dependencies while ensuring working
    logging throughout.

    Returns:
        Configured enhanced configuration repository with all configs loaded

    Raises:
        RuntimeError: If bootstrap fails in an unrecoverable way
    """
    if self._bootstrap_completed and self.config_repository is not None:
        self.logger.info("Bootstrap already completed, returning existing config repository")
        return self.config_repository

    self.logger.info(
        "Starting application bootstrap sequence",
        extra={
            "operation": "bootstrap_start",
            "component": "ApplicationBootstrap",
            "bootstrap_phase": "initialization",
        },
    )

    try:
        # Phase 1: Initial telemetry setup with environment variables only
        self._initialize_telemetry_from_environment()

        # Phase 2: Load enhanced configuration from App Config/Key Vault
        self._load_enhanced_configuration()

        # Phase 3: Upgrade telemetry if App Insights connection string is now available
        self._upgrade_telemetry_from_config()

        # Phase 4: Final configuration loading to os.environ
        self._finalize_configuration_loading()

        # Mark bootstrap as completed
        self._bootstrap_completed = True

        self.logger.info(
            "Application bootstrap completed successfully",
            extra={
                "operation": "bootstrap_complete",
                "component": "ApplicationBootstrap",
                "bootstrap_phase": "completed",
                "telemetry_enabled": bool(telemetry_manager.tracer),
                "config_repository_type": type(self.config_repository).__name__,
            },
        )

        if self.config_repository is not None:
            return self.config_repository
        raise RuntimeError("Bootstrap completed but config repository is None")

    except Exception as e:
        self.logger.error(
            "Application bootstrap failed",
            extra={
                "error_type": type(e).__name__,
                "error_message": str(e),
                "operation": "bootstrap_error",
                "component": "ApplicationBootstrap",
                "bootstrap_phase": "failed",
            },
            exc_info=True,
        )
        raise RuntimeError(f"Application bootstrap failed: {str(e)}") from e

get_config_repository

get_config_repository() -> EnhancedConfigRepositoryInterface | None

Get the configured repository after bootstrap.

Returns:

Type	Description
EnhancedConfigRepositoryInterface | None	The enhanced configuration repository if bootstrap is complete, None otherwise

Source code in azure_bootstrap/services/application_bootstrap.py

def get_config_repository(self) -> EnhancedConfigRepositoryInterface | None:
    """
    Get the configured repository after bootstrap.

    Returns:
        The enhanced configuration repository if bootstrap is complete, None otherwise
    """
    return self.config_repository if self._bootstrap_completed else None

is_bootstrap_completed

is_bootstrap_completed() -> bool

Check if bootstrap process has completed successfully.

Returns:

Type	Description
bool	True if bootstrap completed successfully, False otherwise

Source code in azure_bootstrap/services/application_bootstrap.py

def is_bootstrap_completed(self) -> bool:
    """
    Check if bootstrap process has completed successfully.

    Returns:
        True if bootstrap completed successfully, False otherwise
    """
    return self._bootstrap_completed

BootstrapLogger

Bases: BootstrapLoggerInterface

Bootstrap logging manager that provides safe logging before full configuration.

This class handles the chicken-and-egg problem where: 1. Configuration loading needs logging 2. Logging (App Insights) needs configuration

Solution: Start with basic logging, then upgrade to full telemetry later.

Methods:

Name	Description
configure_bootstrap_logging	Configure basic logging that works before full configuration is loaded.
is_bootstrap_configured	Check if bootstrap logging is configured.
create_logger	Create a logger with bootstrap configuration if not already done.

configure_bootstrap_logging classmethod

configure_bootstrap_logging(level: str | None = None) -> None

Configure basic logging that works before full configuration is loaded.

This provides immediate, safe logging during application bootstrap. Later, telemetry_manager.configure() will enhance it with App Insights.

Parameters:

Name	Type	Description	Default
level	str | None	Logging level (DEBUG, INFO, WARNING, ERROR). If not provided, reads from LOG_LEVEL environment variable, defaulting to INFO.	None

Source code in azure_bootstrap/services/bootstrap_logging.py

@classmethod
def configure_bootstrap_logging(cls, level: str | None = None) -> None:
    """
    Configure basic logging that works before full configuration is loaded.

    This provides immediate, safe logging during application bootstrap.
    Later, telemetry_manager.configure() will enhance it with App Insights.

    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR). If not provided,
               reads from LOG_LEVEL environment variable, defaulting to INFO.
    """
    if cls._configured:
        return

    # Determine logging level: parameter > env var > default (INFO)
    if level is None:
        level = os.environ.get("LOG_LEVEL", "INFO")

    # Validate the level - default to INFO if invalid
    numeric_level = getattr(logging, level.upper(), None)
    if not isinstance(numeric_level, int):
        numeric_level = logging.INFO
        level = "INFO"

    # Set up basic formatter for bootstrap phase with extra fields support
    cls._basic_formatter = ExtraFieldsFormatter(
        "%(asctime)s - [BOOTSTRAP] - %(name)s - %(levelname)s - %(message)s"
    )

    # Configure root logger with basic settings
    root_logger = logging.getLogger()
    root_logger.setLevel(numeric_level)

    # Ensure we have at least one handler
    if not root_logger.handlers:
        handler = logging.StreamHandler()
        if cls._basic_formatter is not None:
            handler.setFormatter(cls._basic_formatter)
        root_logger.addHandler(handler)
    else:
        # Update existing handlers with bootstrap formatter
        if cls._basic_formatter is not None:
            for handler in root_logger.handlers:  # type: ignore
                if hasattr(handler, "setFormatter"):
                    handler.setFormatter(cls._basic_formatter)

    cls._configured = True

    # Log bootstrap completion
    bootstrap_logger = logging.getLogger(__name__)
    bootstrap_logger.info(
        "Bootstrap logging configured successfully",
        extra={"phase": "bootstrap", "level": level, "handlers": len(root_logger.handlers)},
    )

is_bootstrap_configured classmethod

is_bootstrap_configured() -> bool

Check if bootstrap logging is configured.

Source code in azure_bootstrap/services/bootstrap_logging.py

@classmethod
def is_bootstrap_configured(cls) -> bool:
    """Check if bootstrap logging is configured."""
    return cls._configured

create_logger classmethod

create_logger(name: str) -> Logger

Create a logger with bootstrap configuration if not already done.

Parameters:

Name	Type	Description	Default
name	str	Logger name (typically name)	required

Returns:

Type	Description
Logger	Configured logger ready for use

Source code in azure_bootstrap/services/bootstrap_logging.py

@classmethod
def create_logger(cls, name: str) -> logging.Logger:
    """
    Create a logger with bootstrap configuration if not already done.

    Args:
        name: Logger name (typically __name__)

    Returns:
        Configured logger ready for use
    """
    if not cls._configured:
        cls.configure_bootstrap_logging()

    return logging.getLogger(name)

ExtraFieldsFormatter

Bases: Formatter

Custom formatter that appends extra fields from log records to the message.

This ensures that any extra={...} dict passed to logger calls is included in the output, making structured logging data visible in console output.

ApplicationBootstrapInterface

Bases: ABC

Interface for application bootstrap orchestrator.

Defines the contract for handling the complete application startup sequence with proper logging flow.

Methods:

Name	Description
initialize	Execute the complete bootstrap sequence with proper logging flow.
get_config_repository	Get the configured repository after bootstrap.
is_bootstrap_completed	Check if bootstrap process has completed successfully.

initialize abstractmethod

initialize() -> EnhancedConfigRepositoryInterface

Execute the complete bootstrap sequence with proper logging flow.

This method orchestrates the entire application startup process following the correct sequence to avoid circular dependencies while ensuring working logging throughout.

Returns:

Type	Description
EnhancedConfigRepositoryInterface	Configured enhanced configuration repository with all configs loaded

Raises:

Type	Description
RuntimeError	If bootstrap fails in an unrecoverable way

Source code in azure_bootstrap/services/interfaces/application_bootstrap_interface.py

@abstractmethod
def initialize(self) -> "EnhancedConfigRepositoryInterface":
    """
    Execute the complete bootstrap sequence with proper logging flow.

    This method orchestrates the entire application startup process following
    the correct sequence to avoid circular dependencies while ensuring working
    logging throughout.

    Returns:
        Configured enhanced configuration repository with all configs loaded

    Raises:
        RuntimeError: If bootstrap fails in an unrecoverable way
    """
    pass

get_config_repository abstractmethod

get_config_repository() -> Optional[EnhancedConfigRepositoryInterface]

Get the configured repository after bootstrap.

Returns:

Type	Description
Optional[EnhancedConfigRepositoryInterface]	The enhanced configuration repository if bootstrap is complete, None otherwise

Source code in azure_bootstrap/services/interfaces/application_bootstrap_interface.py

@abstractmethod
def get_config_repository(self) -> Optional["EnhancedConfigRepositoryInterface"]:
    """
    Get the configured repository after bootstrap.

    Returns:
        The enhanced configuration repository if bootstrap is complete, None otherwise
    """
    pass

is_bootstrap_completed abstractmethod

is_bootstrap_completed() -> bool

Check if bootstrap process has completed successfully.

Returns:

Type	Description
bool	True if bootstrap completed successfully, False otherwise

Source code in azure_bootstrap/services/interfaces/application_bootstrap_interface.py

@abstractmethod
def is_bootstrap_completed(self) -> bool:
    """
    Check if bootstrap process has completed successfully.

    Returns:
        True if bootstrap completed successfully, False otherwise
    """
    pass

BootstrapLoggerInterface

Bases: ABC

Interface for bootstrap logging manager.

Defines the contract for providing safe logging before full configuration is loaded and App Insights is configured.

Methods:

Name	Description
configure_bootstrap_logging	Configure basic logging that works before full configuration is loaded.
is_bootstrap_configured	Check if bootstrap logging is configured.
create_logger	Create a logger with bootstrap configuration if not already done.

configure_bootstrap_logging abstractmethod classmethod

configure_bootstrap_logging(level: str = 'INFO') -> None

Configure basic logging that works before full configuration is loaded.

This provides immediate, safe logging during application bootstrap. Later, telemetry_manager.configure() will enhance it with App Insights.

Parameters:

Name	Type	Description	Default
level	str	Logging level (DEBUG, INFO, WARNING, ERROR)	'INFO'

Source code in azure_bootstrap/services/interfaces/bootstrap_logger_interface.py

@classmethod
@abstractmethod
def configure_bootstrap_logging(cls, level: str = "INFO") -> None:
    """
    Configure basic logging that works before full configuration is loaded.

    This provides immediate, safe logging during application bootstrap.
    Later, telemetry_manager.configure() will enhance it with App Insights.

    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR)
    """
    pass

is_bootstrap_configured abstractmethod classmethod

is_bootstrap_configured() -> bool

Check if bootstrap logging is configured.

Returns:

Type	Description
bool	True if bootstrap logging is configured, False otherwise

Source code in azure_bootstrap/services/interfaces/bootstrap_logger_interface.py

@classmethod
@abstractmethod
def is_bootstrap_configured(cls) -> bool:
    """
    Check if bootstrap logging is configured.

    Returns:
        True if bootstrap logging is configured, False otherwise
    """
    pass

create_logger abstractmethod classmethod

create_logger(name: str) -> Logger

Create a logger with bootstrap configuration if not already done.

Parameters:

Name	Type	Description	Default
name	str	Logger name (typically name)	required

Returns:

Type	Description
Logger	Configured logger ready for use

Source code in azure_bootstrap/services/interfaces/bootstrap_logger_interface.py

@classmethod
@abstractmethod
def create_logger(cls, name: str) -> logging.Logger:
    """
    Create a logger with bootstrap configuration if not already done.

    Args:
        name: Logger name (typically __name__)

    Returns:
        Configured logger ready for use
    """
    pass

TelemetryManagerInterface

Bases: ABC

Interface for telemetry manager.

Defines the contract for managing Application Insights telemetry and structured logging throughout the application.

Methods:

Name	Description
configure	Configure Application Insights telemetry with support for bootstrap flow reconfiguration.
try_upgrade_from_config	Attempt to upgrade from basic logging to App Insights after config is loaded.
get_tracer	Get the configured tracer.
create_span	Create a new trace span.
log_email_processing_start	Log email processing start with structured data.
log_email_processing_success	Log successful email processing.
log_email_processing_error	Log email processing error.
log_queue_message_received	Log queue message processing.
log_storage_operation	Log storage operations.

configure abstractmethod

configure(connection_string: str | None = None, allow_reconfigure: bool = False) -> bool

Configure Application Insights telemetry with support for bootstrap flow reconfiguration.

Parameters:

Name	Type	Description	Default
connection_string	str | None	App Insights connection string	None
allow_reconfigure	bool	If True, allows reconfiguration even if already configured	False

Returns:

Type	Description
bool	True if configuration succeeded, False otherwise

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def configure(
    self, connection_string: str | None = None, allow_reconfigure: bool = False
) -> bool:
    """
    Configure Application Insights telemetry with support for bootstrap flow reconfiguration.

    Args:
        connection_string: App Insights connection string
        allow_reconfigure: If True, allows reconfiguration even if already configured

    Returns:
        True if configuration succeeded, False otherwise
    """
    pass

try_upgrade_from_config abstractmethod

try_upgrade_from_config(config_repository: EnhancedConfigRepositoryInterface) -> bool

Attempt to upgrade from basic logging to App Insights after config is loaded.

This method is called after configuration loading to check if an App Insights connection string is now available from App Config/Key Vault that wasn't available during initial bootstrap.

Parameters:

Name	Type	Description	Default
config_repository	EnhancedConfigRepositoryInterface	Enhanced config repository to check for connection string	required

Returns:

Type	Description
bool	True if upgrade was attempted (success or failure), False if no upgrade needed

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def try_upgrade_from_config(
    self, config_repository: "EnhancedConfigRepositoryInterface"
) -> bool:
    """
    Attempt to upgrade from basic logging to App Insights after config is loaded.

    This method is called after configuration loading to check if an App Insights
    connection string is now available from App Config/Key Vault that wasn't
    available during initial bootstrap.

    Args:
        config_repository: Enhanced config repository to check for connection string

    Returns:
        True if upgrade was attempted (success or failure), False if no upgrade needed
    """
    pass

get_tracer abstractmethod

get_tracer() -> Any

Get the configured tracer.

Returns:

Type	Description
Any	The OpenTelemetry tracer instance if configured, None otherwise

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def get_tracer(self) -> Any:
    """
    Get the configured tracer.

    Returns:
        The OpenTelemetry tracer instance if configured, None otherwise
    """
    pass

create_span abstractmethod

create_span(name: str, attributes: dict[str, Any] | None = None) -> Any

Create a new trace span.

Parameters:

Name	Type	Description	Default
name	str	Name of the span	required
attributes	dict[str, Any] | None	Optional attributes to attach to the span	None

Returns:

Type	Description
Any	Span instance if telemetry is configured, None otherwise

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def create_span(self, name: str, attributes: dict[str, Any] | None = None) -> Any:
    """
    Create a new trace span.

    Args:
        name: Name of the span
        attributes: Optional attributes to attach to the span

    Returns:
        Span instance if telemetry is configured, None otherwise
    """
    pass

log_email_processing_start abstractmethod

log_email_processing_start(message_id: str | None = None, user_email: str | None = None) -> None

Log email processing start with structured data.

Parameters:

Name	Type	Description	Default
message_id	str | None	Email message ID	None
user_email	str | None	User email address	None

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def log_email_processing_start(
    self, message_id: str | None = None, user_email: str | None = None
) -> None:
    """
    Log email processing start with structured data.

    Args:
        message_id: Email message ID
        user_email: User email address
    """
    pass

log_email_processing_success abstractmethod

log_email_processing_success(message_id: str, user_email: str, processing_time_ms: int) -> None

Log successful email processing.

Parameters:

Name	Type	Description	Default
message_id	str	Email message ID	required
user_email	str	User email address	required
processing_time_ms	int	Processing time in milliseconds	required

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def log_email_processing_success(
    self, message_id: str, user_email: str, processing_time_ms: int
) -> None:
    """
    Log successful email processing.

    Args:
        message_id: Email message ID
        user_email: User email address
        processing_time_ms: Processing time in milliseconds
    """
    pass

log_email_processing_error abstractmethod

log_email_processing_error(error: str, message_id: str | None = None, user_email: str | None = None) -> None

Log email processing error.

Parameters:

Name	Type	Description	Default
error	str	Error message	required
message_id	str | None	Email message ID (if available)	None
user_email	str | None	User email address (if available)	None

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def log_email_processing_error(
    self, error: str, message_id: str | None = None, user_email: str | None = None
) -> None:
    """
    Log email processing error.

    Args:
        error: Error message
        message_id: Email message ID (if available)
        user_email: User email address (if available)
    """
    pass

log_queue_message_received abstractmethod

log_queue_message_received(queue_name: str, message_id: str) -> None

Log queue message processing.

Parameters:

Name	Type	Description	Default
queue_name	str	Name of the queue	required
message_id	str	Message ID	required

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def log_queue_message_received(self, queue_name: str, message_id: str) -> None:
    """
    Log queue message processing.

    Args:
        queue_name: Name of the queue
        message_id: Message ID
    """
    pass

log_storage_operation abstractmethod

log_storage_operation(operation: str, container: str, blob_name: str, success: bool) -> None

Log storage operations.

Parameters:

Name	Type	Description	Default
operation	str	Type of storage operation	required
container	str	Storage container name	required
blob_name	str	Blob name	required
success	bool	Whether the operation succeeded	required

Source code in azure_bootstrap/services/interfaces/telemetry_manager_interface.py

@abstractmethod
def log_storage_operation(
    self, operation: str, container: str, blob_name: str, success: bool
) -> None:
    """
    Log storage operations.

    Args:
        operation: Type of storage operation
        container: Storage container name
        blob_name: Blob name
        success: Whether the operation succeeded
    """
    pass

TelemetryManager

TelemetryManager()

Bases: TelemetryManagerInterface

Manages Application Insights telemetry and structured logging

Methods:

Name	Description
configure	Configure Application Insights telemetry with support for bootstrap flow reconfiguration
try_upgrade_from_config	Attempt to upgrade from basic logging to App Insights after config is loaded
get_tracer	Get the configured tracer
create_span	Create a new trace span
log_email_processing_start	Log email processing start with structured data
log_email_processing_success	Log successful email processing
log_email_processing_error	Log email processing error
log_queue_message_received	Log queue message processing
log_storage_operation	Log storage operations

Source code in azure_bootstrap/services/telemetry.py

def __init__(self) -> None:
    self.tracer: Any = None
    self._configured = False

configure

configure(connection_string: str | None = None, allow_reconfigure: bool = False) -> bool

Configure Application Insights telemetry with support for bootstrap flow reconfiguration

Parameters:

Name	Type	Description	Default
connection_string	str | None	App Insights connection string	None
allow_reconfigure	bool	If True, allows reconfiguration even if already configured	False

Source code in azure_bootstrap/services/telemetry.py

def configure(
    self, connection_string: str | None = None, allow_reconfigure: bool = False
) -> bool:
    """
    Configure Application Insights telemetry with support for bootstrap flow reconfiguration

    Args:
        connection_string: App Insights connection string
        allow_reconfigure: If True, allows reconfiguration even if already configured
    """
    if self._configured and not allow_reconfigure:
        return True

    try:
        # Get connection string from environment or parameter
        app_insights_connection_string = connection_string or os.environ.get(
            "APPLICATIONINSIGHTS_CONNECTION_STRING"
        )

        if not app_insights_connection_string:
            logging.warning(
                "Application Insights connection string not found. Using basic logging."
            )
            self._configure_logging()
            self._configured = True
            return True

        if not TELEMETRY_AVAILABLE:
            logging.warning("Azure Monitor OpenTelemetry not available. Using basic logging.")
            self._configure_logging()
            self._configured = True
            return True

        # Configure Azure Monitor
        configure_azure_monitor(
            connection_string=app_insights_connection_string,
            enable_live_metrics=True,
        )

        # Instrument Azure Functions (if available)
        if AZURE_FUNCTIONS_INSTRUMENTOR_AVAILABLE and AzureFunctionsInstrumentor:
            AzureFunctionsInstrumentor().instrument()

        # Get tracer
        self.tracer = trace.get_tracer(__name__)

        # Configure structured logging
        self._configure_logging()

        self._configured = True
        logging.info("Application Insights telemetry configured successfully")
        return True

    except Exception as e:
        logging.error(f"Failed to configure Application Insights: {str(e)}")
        # Fallback to basic logging
        self._configure_logging()
        self._configured = True
        return True

try_upgrade_from_config

try_upgrade_from_config(config_repository: EnhancedConfigRepositoryInterface) -> bool

Attempt to upgrade from basic logging to App Insights after config is loaded

This method is called after configuration loading to check if an App Insights connection string is now available from App Config/Key Vault that wasn't available during initial bootstrap.

Parameters:

Name	Type	Description	Default
config_repository	EnhancedConfigRepositoryInterface	Enhanced config repository to check for connection string	required

Returns:

Type	Description
bool	True if upgrade was attempted (success or failure), False if no upgrade needed

Source code in azure_bootstrap/services/telemetry.py

def try_upgrade_from_config(self, config_repository: EnhancedConfigRepositoryInterface) -> bool:
    """
    Attempt to upgrade from basic logging to App Insights after config is loaded

    This method is called after configuration loading to check if an App Insights
    connection string is now available from App Config/Key Vault that wasn't
    available during initial bootstrap.

    Args:
        config_repository: Enhanced config repository to check for connection string

    Returns:
        True if upgrade was attempted (success or failure), False if no upgrade needed
    """
    # Don't upgrade if already using App Insights
    if self.tracer is not None:
        logging.debug("Already using Application Insights, no upgrade needed")
        return False

    # Check if connection string is now available from config
    # First try get_value() which checks env vars, cache, App Config, then Key Vault fallback
    # If not found, try get_secret_value() for direct Key Vault access
    try:
        app_insights_connection_string = config_repository.get_value(
            "APPLICATIONINSIGHTS_CONNECTION_STRING"
        )

        # If not found via get_value, try direct Key Vault access as last resort
        if not app_insights_connection_string:
            app_insights_connection_string = config_repository.get_secret_value(
                "APPLICATIONINSIGHTS_CONNECTION_STRING"
            )

        if app_insights_connection_string:
            logging.info(
                "Found Application Insights connection string in config, upgrading from basic logging"
            )

            # Reconfigure with the new connection string
            success = self.configure(
                connection_string=app_insights_connection_string, allow_reconfigure=True
            )

            if success and self.tracer:
                logging.info("Successfully upgraded to Application Insights telemetry")
                return True
            else:
                logging.warning(
                    "Failed to upgrade to Application Insights, continuing with basic logging"
                )
                return True
        else:
            logging.debug("No Application Insights connection string found in config")
            return False

    except Exception as e:
        logging.warning(f"Error checking for Application Insights upgrade: {str(e)}")
        return False

get_tracer

get_tracer() -> Any

Get the configured tracer

Source code in azure_bootstrap/services/telemetry.py

def get_tracer(self) -> Any:
    """Get the configured tracer"""
    return self.tracer

create_span

create_span(name: str, attributes: dict[str, Any] | None = None) -> Any

Create a new trace span

Source code in azure_bootstrap/services/telemetry.py

def create_span(self, name: str, attributes: dict[str, Any] | None = None) -> Any:
    """Create a new trace span"""
    # Check if telemetry is available and tracer is configured
    has_tracer = bool(getattr(self, "tracer", None))
    if TELEMETRY_AVAILABLE and has_tracer and self.tracer is not None:
        return self.tracer.start_span(name, attributes=attributes)
    return None

log_email_processing_start

log_email_processing_start(message_id: str | None = None, user_email: str | None = None) -> None

Log email processing start with structured data

Source code in azure_bootstrap/services/telemetry.py

def log_email_processing_start(
    self, message_id: str | None = None, user_email: str | None = None
) -> None:
    """Log email processing start with structured data"""
    log_data = {
        "event": "email_processing_start",
        "message_id": message_id,
        "user_email": user_email,
        "operation": "read_email",
    }
    logging.info("Email processing started", extra=log_data)

log_email_processing_success

log_email_processing_success(message_id: str, user_email: str, processing_time_ms: int) -> None

Log successful email processing

Source code in azure_bootstrap/services/telemetry.py

def log_email_processing_success(
    self, message_id: str, user_email: str, processing_time_ms: int
) -> None:
    """Log successful email processing"""
    log_data = {
        "event": "email_processing_success",
        "message_id": message_id,
        "user_email": user_email,
        "processing_time_ms": processing_time_ms,
        "operation": "read_email",
    }
    logging.info("Email processing completed successfully", extra=log_data)

log_email_processing_error

log_email_processing_error(error: str, message_id: str | None = None, user_email: str | None = None) -> None

Log email processing error

Source code in azure_bootstrap/services/telemetry.py

def log_email_processing_error(
    self, error: str, message_id: str | None = None, user_email: str | None = None
) -> None:
    """Log email processing error"""
    log_data = {
        "event": "email_processing_error",
        "error": error,
        "message_id": message_id,
        "user_email": user_email,
        "operation": "read_email",
    }
    logging.error("Email processing failed", extra=log_data)

log_queue_message_received

log_queue_message_received(queue_name: str, message_id: str) -> None

Log queue message processing

Source code in azure_bootstrap/services/telemetry.py

def log_queue_message_received(self, queue_name: str, message_id: str) -> None:
    """Log queue message processing"""
    log_data = {
        "event": "queue_message_received",
        "queue_name": queue_name,
        "message_id": message_id,
        "operation": "queue_processing",
    }
    logging.info("Queue message received", extra=log_data)

log_storage_operation

log_storage_operation(operation: str, container: str, blob_name: str, success: bool) -> None

Log storage operations

Source code in azure_bootstrap/services/telemetry.py

def log_storage_operation(
    self, operation: str, container: str, blob_name: str, success: bool
) -> None:
    """Log storage operations"""
    log_data = {
        "event": "storage_operation",
        "operation": operation,
        "container": container,
        "blob_name": blob_name,
        "success": success,
    }
    level = logging.INFO if success else logging.ERROR
    logging.log(level, f"Storage operation: {operation}", extra=log_data)

build_info

build_info() -> dict[str, str | None]

Return version/build metadata from downward-API or CI env vars.

Source code in azure_bootstrap/aks/__init__.py

def build_info() -> dict[str, str | None]:
    """Return version/build metadata from downward-API or CI env vars."""
    return {
        "version": os.environ.get("BUILD_VERSION") or os.environ.get("APP_VERSION"),
        "git_sha": os.environ.get("GIT_SHA"),
        "build_time": os.environ.get("BUILD_TIME"),
        "image_tag": os.environ.get("IMAGE_TAG"),
        "pod_name": os.environ.get("POD_NAME"),
        "pod_namespace": os.environ.get("POD_NAMESPACE"),
        "node_name": os.environ.get("NODE_NAME"),
    }

verify_chain

verify_chain(records: list[ChainedAuditRecord]) -> bool

Verify the integrity of an ordered list of :class:ChainedAuditRecord objects.

For each record the function recomputes the expected record_hash from the record's fields and confirms it matches the stored record_hash. It also verifies that each record's prev_hash equals the preceding record's record_hash (or None for the first record).

Returns True if the chain is intact, False on the first detected anomaly (and logs a warning with tamper-evidence details).

Parameters

records: Ordered list of records as originally produced by :meth:AuditChain.append_chained.

Source code in azure_bootstrap/audit/__init__.py

def verify_chain(records: list[ChainedAuditRecord]) -> bool:
    """Verify the integrity of an ordered list of :class:`ChainedAuditRecord` objects.

    For each record the function recomputes the expected ``record_hash`` from
    the record's fields and confirms it matches the stored ``record_hash``.  It
    also verifies that each record's ``prev_hash`` equals the preceding record's
    ``record_hash`` (or ``None`` for the first record).

    Returns ``True`` if the chain is intact, ``False`` on the first detected
    anomaly (and logs a warning with tamper-evidence details).

    Parameters
    ----------
    records:
        Ordered list of records as originally produced by
        :meth:`AuditChain.append_chained`.
    """
    if not records:
        return True

    expected_prev: str | None = None
    for idx, record in enumerate(records):
        # Verify linkage
        if record.prev_hash != expected_prev:
            _log.warning(
                "audit chain tamper detected: prev_hash mismatch",
                extra={
                    "chain_index": idx,
                    "record_id": record.id,
                    "expected_prev_hash": expected_prev,
                    "actual_prev_hash": record.prev_hash,
                    "event_type": record.event_type,
                    "actor": record.actor,
                    "resource": record.resource,
                },
            )
            bump_counter("audit.chain.tamper_detected")
            return False

        # Verify self-hash
        expected_hash = _compute_record_hash(
            id=record.id,
            ts=record.ts,
            event_type=record.event_type,
            actor=record.actor,
            resource=record.resource,
            detail=record.detail,
            prev_hash=record.prev_hash,
        )
        if record.record_hash != expected_hash:
            _log.warning(
                "audit chain tamper detected: record_hash mismatch",
                extra={
                    "chain_index": idx,
                    "record_id": record.id,
                    "expected_record_hash": expected_hash,
                    "actual_record_hash": record.record_hash,
                    "event_type": record.event_type,
                    "actor": record.actor,
                    "resource": record.resource,
                },
            )
            bump_counter("audit.chain.tamper_detected")
            return False

        expected_prev = record.record_hash

    bump_counter("audit.chain.verify_ok")
    return True

verify_hmac_signature

verify_hmac_signature(secret: str, raw_body: bytes, header_value: str, *, prefix: str = 'sha256=') -> bool

Constant-time HMAC-SHA256 verify (GitHub/Sumo sha256=… style).

Parameters

secret: Shared signing secret. raw_body: Raw request body bytes (must not be re-serialized JSON). header_value: Value of the signature header (may include sha256= prefix). prefix: Expected algorithm prefix in the header value.

Source code in azure_bootstrap/auth/hmac.py

def verify_hmac_signature(
    secret: str,
    raw_body: bytes,
    header_value: str,
    *,
    prefix: str = "sha256=",
) -> bool:
    """Constant-time HMAC-SHA256 verify (GitHub/Sumo ``sha256=…`` style).

    Parameters
    ----------
    secret:
        Shared signing secret.
    raw_body:
        Raw request body bytes (must not be re-serialized JSON).
    header_value:
        Value of the signature header (may include ``sha256=`` prefix).
    prefix:
        Expected algorithm prefix in the header value.
    """
    if not secret or not header_value:
        return False
    provided = header_value.strip()
    if provided.startswith(prefix):
        provided = provided[len(prefix) :]
    elif "=" in provided:
        provided = provided.split("=", 1)[1]
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, provided.lower())

bootstrap_initialized

bootstrap_initialized() -> bool

Process-local state probe — useful for /api/health/ready endpoints.

Source code in azure_bootstrap/bootstrap/__init__.py

def bootstrap_initialized() -> bool:
    """Process-local state probe — useful for /api/health/ready endpoints."""
    return _bootstrap_initialized

ensure_bootstrap

ensure_bootstrap() -> None

Lazy idempotent wrapper around v1 initialize_application.

Short-circuits when USE_MOCK_BOOTSTRAP is truthy (sets the initialized flag and returns without contacting Azure). On real bootstrap, calls v1 initialize_application() and sets the flag. Re-raises on failure after logging at ERROR with traceback.

Source code in azure_bootstrap/bootstrap/__init__.py

def ensure_bootstrap() -> None:
    """Lazy idempotent wrapper around v1 ``initialize_application``.

    Short-circuits when ``USE_MOCK_BOOTSTRAP`` is truthy (sets the
    initialized flag and returns without contacting Azure). On real
    bootstrap, calls v1 ``initialize_application()`` and sets the flag.
    Re-raises on failure after logging at ERROR with traceback.
    """
    global _bootstrap_initialized
    if _bootstrap_initialized:
        return
    logger = get_bootstrap_logger(__name__)
    if _mock_bootstrap_enabled():
        logger.info("Bootstrap mock mode active; skipping Azure connections")
        _bootstrap_initialized = True
        return
    try:
        from azure_bootstrap.services.application_bootstrap import initialize_application

        initialize_application()
        _bootstrap_initialized = True
    except Exception:
        logger.error("Bootstrap initialization failed", exc_info=True)
        raise

load_local_settings

load_local_settings(path: str | Path = 'local.settings.json') -> int

Load env vars from an Azure-Functions-style local.settings.json.

Behavior: - Missing file: silent, returns 0. - Keys starting with _ are documentation sentinels; skipped. - Never overrides an existing os.environ entry. - JSON / OS errors are logged at WARNING and return 0.

Source code in azure_bootstrap/bootstrap/__init__.py

def load_local_settings(path: str | Path = "local.settings.json") -> int:
    """Load env vars from an Azure-Functions-style ``local.settings.json``.

    Behavior:
    - Missing file: silent, returns 0.
    - Keys starting with ``_`` are documentation sentinels; skipped.
    - Never overrides an existing ``os.environ`` entry.
    - JSON / OS errors are logged at WARNING and return 0.
    """
    p = Path(path)
    if not p.exists():
        return 0
    logger = logging.getLogger(__name__)
    try:
        raw = p.read_text(encoding="utf-8")
        data = json.loads(raw)
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("Failed to load local settings from %s: %s", path, exc)
        return 0
    values = data.get("Values") if isinstance(data, dict) else None
    if not isinstance(values, dict):
        return 0
    loaded = 0
    for key, value in values.items():
        if not isinstance(key, str) or key.startswith("_"):
            continue
        if key in os.environ:
            continue
        try:
            os.environ[key] = str(value)
            loaded += 1
        except Exception:
            continue
    return loaded

bump_counter

bump_counter(name: str, n: int = 1) -> None

Thread-safe increment. Never raises.

Source code in azure_bootstrap/counters/__init__.py

def bump_counter(name: str, n: int = 1) -> None:
    """Thread-safe increment. Never raises."""
    if not isinstance(name, str) or not name:
        return
    try:
        with _lock:
            _counters[name] = _counters.get(name, 0) + int(n)
    except Exception:
        pass

counter_snapshot

counter_snapshot() -> dict[str, int]

Return a copy of the counter map.

Source code in azure_bootstrap/counters/__init__.py

def counter_snapshot() -> dict[str, int]:
    """Return a copy of the counter map."""
    with _lock:
        return dict(_counters)

drain_outbox

drain_outbox(session: Any, sender_fn: Callable[[dict[str, Any]], None], *, batch_size: int = 10, max_attempts: int = 5, table: str = 'outbox') -> int

Claim and drain pending outbox rows. Returns count sent.

Source code in azure_bootstrap/db/outbox.py

def drain_outbox(
    session: Any,
    sender_fn: Callable[[dict[str, Any]], None],
    *,
    batch_size: int = 10,
    max_attempts: int = 5,
    table: str = "outbox",
) -> int:
    """Claim and drain pending outbox rows. Returns count sent."""
    from sqlalchemy import text  # type: ignore[import-untyped]

    _validate_identifier(table)
    sql = text(f"""
        SELECT id, payload FROM {table}
        WHERE status = :pending AND attempt_count < :max
        ORDER BY created_at
        LIMIT :batch
        FOR UPDATE SKIP LOCKED
        """)  # nosec B608 — table validated above
    rows = session.execute(
        sql, {"pending": STATUS_PENDING, "max": max_attempts, "batch": batch_size}
    ).fetchall()
    sent = 0
    outbox = Outbox(session, table=table)
    for row in rows:
        msg_id = str(row[0])
        payload = json.loads(row[1]) if isinstance(row[1], str) else row[1]
        if not outbox.claim(msg_id):
            continue
        try:
            sender_fn(payload)
            outbox.mark_sent(msg_id)
            sent += 1
        except Exception as exc:
            outbox.mark_failed(msg_id, str(exc), max_attempts=max_attempts)
            _logger.warning("outbox drain failed for %s: %s", msg_id, exc)
    return sent

is_unrecoverable

is_unrecoverable(exc: BaseException) -> bool

The consumer's classifier — single isinstance check.

Source code in azure_bootstrap/exceptions/__init__.py

def is_unrecoverable(exc: BaseException) -> bool:
    """The consumer's classifier — single ``isinstance`` check."""
    return isinstance(exc, DEFAULT_UNRECOVERABLE_TYPES)

build_session

build_session(*, total_retries: int = 5, backoff_factor: float = 1.0, pool_connections: int = 10, pool_maxsize: int = 10) -> Any

Return a requests.Session with urllib3 Retry mounted.

Source code in azure_bootstrap/http/__init__.py

def build_session(
    *,
    total_retries: int = 5,
    backoff_factor: float = 1.0,
    pool_connections: int = 10,
    pool_maxsize: int = 10,
) -> Any:
    """Return a ``requests.Session`` with urllib3 Retry mounted."""
    import requests  # type: ignore[import-untyped]
    from requests.adapters import HTTPAdapter  # type: ignore[import-untyped]
    from urllib3.util.retry import Retry

    retry = Retry(
        total=total_retries,
        backoff_factor=backoff_factor,
        backoff_jitter=0.3,
        status_forcelist=(408, 429, 500, 502, 503, 504),
        allowed_methods=frozenset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"]),
        respect_retry_after_header=True,
        raise_on_status=False,
    )
    session = requests.Session()
    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize,
    )
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

request_with_retry

request_with_retry(method: str, url: str, *, session: Any | None = None, timeout: float = _DEFAULT_TIMEOUT, headers: dict[str, str] | None = None, allow_private: bool = False, **kwargs: Any) -> Any

Issue an HTTP request with default timeout, traceparent, and SSRF guard.

Source code in azure_bootstrap/http/__init__.py

def request_with_retry(
    method: str,
    url: str,
    *,
    session: Any | None = None,
    timeout: float = _DEFAULT_TIMEOUT,
    headers: dict[str, str] | None = None,
    allow_private: bool = False,
    **kwargs: Any,
) -> Any:
    """Issue an HTTP request with default timeout, traceparent, and SSRF guard."""
    check_ssrf(url, allow_private=allow_private)
    sess = session or build_session()
    hdrs = inject_traceparent(headers)
    return sess.request(method.upper(), url, headers=hdrs, timeout=timeout, **kwargs)

build_tenant_credential

build_tenant_credential(tenant_id: str, *, app_client_id: str | None = None, token_file_path: str = _DEFAULT_TOKEN_FILE) -> Any

Build a WorkloadIdentityCredential scoped to tenant_id.

Zero-secret — uses the federated token file on disk. Designed for multi-tenant Entra applications that need to acquire per-customer-tenant Graph (or ARM) tokens from a single registered app.

Parameters:

Name	Type	Description	Default
tenant_id	str	The target customer / resource tenant.	required
app_client_id	str | None	The multi-tenant app's client ID. Falls back to AZURE_CLIENT_ID env if not supplied.	None
token_file_path	str	Path to the federated token file written by the Azure Workload Identity webhook. Defaults to the standard AKS path.	_DEFAULT_TOKEN_FILE

Returns:

Type	Description
Any	A WorkloadIdentityCredential instance.

Raises:

Type	Description
ValueError	If no client ID can be resolved.
ImportError	If azure-identity is not installed.

Source code in azure_bootstrap/identity/__init__.py

def build_tenant_credential(
    tenant_id: str,
    *,
    app_client_id: str | None = None,
    token_file_path: str = _DEFAULT_TOKEN_FILE,
) -> Any:
    """Build a ``WorkloadIdentityCredential`` scoped to *tenant_id*.

    Zero-secret — uses the federated token file on disk.  Designed for
    multi-tenant Entra applications that need to acquire per-customer-tenant
    Graph (or ARM) tokens from a single registered app.

    Args:
        tenant_id: The target customer / resource tenant.
        app_client_id: The multi-tenant app's client ID.  Falls back to
            ``AZURE_CLIENT_ID`` env if not supplied.
        token_file_path: Path to the federated token file written by the
            Azure Workload Identity webhook.  Defaults to the standard
            AKS path.

    Returns:
        A ``WorkloadIdentityCredential`` instance.

    Raises:
        ValueError: If no client ID can be resolved.
        ImportError: If ``azure-identity`` is not installed.
    """
    try:
        from azure.identity import WorkloadIdentityCredential  # type: ignore[import-not-found]
    except ImportError as exc:  # pragma: no cover
        raise ImportError("build_tenant_credential requires azure-identity (in core deps)") from exc

    client = app_client_id or os.environ.get("AZURE_CLIENT_ID", "").strip() or None
    if not client:
        raise ValueError(
            "build_tenant_credential requires app_client_id or AZURE_CLIENT_ID env var"
        )

    _logger.info(
        "Tenant credential built",
        extra={
            "operation": "identity.build_tenant_credential",
            "tenant_id": tenant_id,
            "client_id": client,
            "token_file_path": token_file_path,
        },
    )
    bump_counter("identity.credential_built.tenant_workload_identity")
    return WorkloadIdentityCredential(
        tenant_id=tenant_id,
        client_id=client,
        token_file_path=token_file_path,
    )

build_tenant_credential_cached

build_tenant_credential_cached(tenant_id: str, scope: str, *, app_client_id: str | None = None) -> str

Return an access token for tenant_id / scope, using the cache.

On a cache hit the cached token string is returned immediately. On a miss a fresh WorkloadIdentityCredential is built via :func:build_tenant_credential, get_token(scope) is called, and the result is stored in the cache before being returned.

Parameters:

Name	Type	Description	Default
tenant_id	str	The target customer / resource tenant.	required
scope	str	The OAuth2 scope to request (e.g. "https://graph.microsoft.com/.default").	required
app_client_id	str | None	Passed through to :func:build_tenant_credential.	None

Returns:

Type	Description
str	The raw token string (AccessToken.token).

Raises:

Type	Description
ValueError	If no client ID can be resolved.
ImportError	If azure-identity is not installed.

Source code in azure_bootstrap/identity/__init__.py

def build_tenant_credential_cached(
    tenant_id: str,
    scope: str,
    *,
    app_client_id: str | None = None,
) -> str:
    """Return an access token for *tenant_id* / *scope*, using the cache.

    On a cache hit the cached token string is returned immediately.
    On a miss a fresh ``WorkloadIdentityCredential`` is built via
    :func:`build_tenant_credential`, ``get_token(scope)`` is called, and
    the result is stored in the cache before being returned.

    Args:
        tenant_id: The target customer / resource tenant.
        scope: The OAuth2 scope to request (e.g.
            ``"https://graph.microsoft.com/.default"``).
        app_client_id: Passed through to :func:`build_tenant_credential`.

    Returns:
        The raw token string (``AccessToken.token``).

    Raises:
        ValueError: If no client ID can be resolved.
        ImportError: If ``azure-identity`` is not installed.
    """
    cached = TokenCache.get_cached_token(tenant_id, scope)
    if cached is not None:
        bump_counter("identity.token_cache.hit")
        return cached

    bump_counter("identity.token_cache.miss")
    cred = build_tenant_credential(
        tenant_id,
        app_client_id=app_client_id,
    )
    access_token = cred.get_token(scope)
    token_str: str = access_token.token
    expires_at: float = float(getattr(access_token, "expires_on", 0))
    TokenCache.cache_token(tenant_id, scope, token_str, expires_at)
    return token_str

configure_logging

configure_logging(*, format_string: str = '%(asctime)s %(levelname)s %(name)s %(message)s', silence_defaults: bool = True, extra_noisy_loggers: tuple[str, ...] = ()) -> None

Install structured-logging defaults. Idempotent — replaces handlers.

Source code in azure_bootstrap/logging/config.py

def configure_logging(
    *,
    format_string: str = "%(asctime)s %(levelname)s %(name)s %(message)s",
    silence_defaults: bool = True,
    extra_noisy_loggers: tuple[str, ...] = (),
) -> None:
    """Install structured-logging defaults. Idempotent — replaces handlers."""
    logging.setLoggerClass(_StrictLogger)

    level = effective_log_level()
    handler = logging.StreamHandler()
    handler.setFormatter(ExtraFieldsFormatter(format_string))
    handler.addFilter(CorrelationFilter())

    logging.basicConfig(level=level, handlers=[handler], force=True)

    root = logging.getLogger()
    if not any(isinstance(f, CorrelationFilter) for f in root.filters):
        root.addFilter(CorrelationFilter())

    silence_noisy_loggers(*extra_noisy_loggers, include_defaults=silence_defaults)

correlation_scope

correlation_scope(correlation_id: str | None = None, **fields: str | None) -> Generator[str, None, None]

Push correlation context for the duration of the with-block.

Yields the resolved correlation_id, always a non-empty string. A fresh 12-char uuid hex is minted when correlation_id is None. Any keyword argument becomes a context var (e.g. email_id, request_id).

Source code in azure_bootstrap/logging/correlation.py

@contextmanager
def correlation_scope(
    correlation_id: str | None = None,
    **fields: str | None,
) -> Generator[str, None, None]:
    """Push correlation context for the duration of the with-block.

    Yields the resolved correlation_id, always a non-empty string. A fresh
    12-char uuid hex is minted when ``correlation_id`` is None. Any keyword
    argument becomes a context var (e.g. ``email_id``, ``request_id``).
    """
    resolved = correlation_id or uuid.uuid4().hex[:12]
    tokens: list[tuple[contextvars.ContextVar[str | None], contextvars.Token[str | None]]] = []
    tokens.append((_var_for("correlation_id"), _var_for("correlation_id").set(resolved)))
    for key, value in fields.items():
        if value is None:
            continue
        var = _var_for(key)
        tokens.append((var, var.set(value)))
    try:
        yield resolved
    finally:
        for var, token in reversed(tokens):
            try:
                var.reset(token)
            except Exception:
                pass

mask_secrets_in_dict

mask_secrets_in_dict(d: dict[str, Any]) -> dict[str, Any]

Shallow copy with secret-keyed values replaced by '***'.

A value is replaced only when truthy — empty strings, None, 0, etc. pass through unmodified so the caller can still see "field was empty".

Source code in azure_bootstrap/logging/masking.py

def mask_secrets_in_dict(d: dict[str, Any]) -> dict[str, Any]:
    """Shallow copy with secret-keyed values replaced by '***'.

    A value is replaced only when truthy — empty strings, None, 0, etc. pass
    through unmodified so the caller can still see "field was empty".
    """
    out: dict[str, Any] = {}
    for key, value in d.items():
        if isinstance(key, str) and key.lower() in _SECRET_KEY_ALLOWLIST and value:
            out[key] = "***"
        else:
            out[key] = value
    return out

sanitize_for_log

sanitize_for_log(value: str | None, *, max_len: int = 256) -> str

Replace control chars (\x00-\x1f, \x7f) with '?' and truncate.

Source code in azure_bootstrap/logging/masking.py

def sanitize_for_log(value: str | None, *, max_len: int = 256) -> str:
    """Replace control chars (\\x00-\\x1f, \\x7f) with '?' and truncate."""
    if value is None:
        return ""
    cleaned = _CONTROL_CHARS_FOR_LOGS.sub("?", value)
    if len(cleaned) > max_len:
        return cleaned[:max_len] + "...[truncated]"
    return cleaned

confine_to_root

confine_to_root(raw: str | Path, *, allowed_root: str | Path) -> Path

Resolve raw and assert it's under allowed_root.

Both inputs are canonicalized via Path.expanduser().resolve() before comparison — protects against symlink escape and .. traversal.

Source code in azure_bootstrap/path_safety/__init__.py

def confine_to_root(raw: str | Path, *, allowed_root: str | Path) -> Path:
    """Resolve ``raw`` and assert it's under ``allowed_root``.

    Both inputs are canonicalized via ``Path.expanduser().resolve()`` before
    comparison — protects against symlink escape and ``..`` traversal.
    """
    try:
        root = Path(allowed_root).expanduser().resolve()
    except (OSError, RuntimeError) as exc:
        raise ValueError(f"allowed_root could not be resolved: {exc}") from exc
    try:
        candidate = Path(raw).expanduser().resolve()
    except (OSError, RuntimeError) as exc:
        raise ValueError(f"path could not be resolved: {exc}") from exc

    try:
        candidate.relative_to(root)
    except ValueError as exc:
        raise ValueError(f"path {str(candidate)!r} escapes allowed root {str(root)!r}") from exc
    return candidate

sanitize_path_segment

sanitize_path_segment(value: str, *, empty_placeholder: str = 'attachment') -> str

Normalize a single filename / path segment.

Order matters: bidi/zero-width chars are stripped BEFORE any other normalization so a visual-spoofing attack can't hide other transformations.

Source code in azure_bootstrap/path_safety/__init__.py

def sanitize_path_segment(value: str, *, empty_placeholder: str = "attachment") -> str:
    """Normalize a single filename / path segment.

    Order matters: bidi/zero-width chars are stripped BEFORE any other
    normalization so a visual-spoofing attack can't hide other transformations.
    """
    if not isinstance(value, str):
        return empty_placeholder
    s = value.strip()
    if not s:
        return empty_placeholder

    s = _BIDI_AND_INVISIBLE_CHARS.sub("", s)
    s = _WHITESPACE_RUN.sub("_", s)
    s = s.replace("-", "_")
    s = _FORBIDDEN_FILENAME_CHARS.sub("_", s)
    s = s.replace("..", "_")
    s = _UNDERSCORE_RUN.sub("_", s)
    s = s.strip("_")
    if len(s) > MAX_SEGMENT_LEN:
        s = s[:MAX_SEGMENT_LEN].rstrip("_")
    return s or empty_placeholder

run_phase

run_phase(name: str, fn: Callable[..., T], *args: Any, namespace: str = 'phase', alert_severity: str | None = 'error', aggregate_counter: str | None = None, **kwargs: Any) -> PhaseResult[T]

Run fn under a try/except. NEVER re-raises.

Counters bumped (best-effort): - {namespace}.{name}.ok on success - {namespace}.{name}.failed on exception - {namespace}.{name}.{aggregate_counter} += len(result) when aggregate_counter is given AND the result supports len.

Source code in azure_bootstrap/phases/__init__.py

def run_phase(
    name: str,
    fn: Callable[..., T],
    *args: Any,
    namespace: str = "phase",
    alert_severity: str | None = "error",
    aggregate_counter: str | None = None,
    **kwargs: Any,
) -> PhaseResult[T]:
    """Run ``fn`` under a try/except. NEVER re-raises.

    Counters bumped (best-effort):
      - ``{namespace}.{name}.ok`` on success
      - ``{namespace}.{name}.failed`` on exception
      - ``{namespace}.{name}.{aggregate_counter} += len(result)`` when
        ``aggregate_counter`` is given AND the result supports ``len``.
    """
    start = time.monotonic()
    try:
        value = fn(*args, **kwargs)
    except BaseException as exc:
        elapsed = round(time.monotonic() - start, 3)
        _logger.exception(
            "Phase %s raised; contributing nothing",
            name,
            extra={
                "phase": name,
                "namespace": namespace,
                "exception_type": type(exc).__name__,
                "elapsed_seconds": elapsed,
            },
        )
        bump_counter(f"{namespace}.{name}.failed")
        if alert_severity:
            _fire_alert(alert_severity, namespace, name, exc)
        return PhaseResult(name=name, ok=False, value=None, exception=exc, elapsed_seconds=elapsed)

    elapsed = round(time.monotonic() - start, 3)
    bump_counter(f"{namespace}.{name}.ok")
    if aggregate_counter is not None:
        try:
            n = len(value)  # type: ignore[arg-type]
            bump_counter(f"{namespace}.{name}.{aggregate_counter}", n)
        except TypeError:
            pass
    return PhaseResult(name=name, ok=True, value=value, elapsed_seconds=elapsed)

run_phases

run_phases(phases: list[tuple[str, Callable[..., Any]]], *, namespace: str = 'phase', alert_severity: str | None = 'error') -> list[PhaseResult[Any]]

Run a list of (name, callable) pairs in order, swallowing per-phase failures so subsequent phases still execute.

Source code in azure_bootstrap/phases/__init__.py

def run_phases(
    phases: list[tuple[str, Callable[..., Any]]],
    *,
    namespace: str = "phase",
    alert_severity: str | None = "error",
) -> list[PhaseResult[Any]]:
    """Run a list of (name, callable) pairs in order, swallowing per-phase
    failures so subsequent phases still execute."""
    return [
        run_phase(name, fn, namespace=namespace, alert_severity=alert_severity)
        for name, fn in phases
    ]

create_enhanced_config_repository

create_enhanced_config_repository(app_config_connection_string: str | None = None, secrets_repository: SecretsRepositoryInterface | None = None, auto_load_to_environ: bool = False) -> EnhancedConfigRepositoryInterface

Factory function to create an enhanced configuration repository.

This factory function provides a convenient way to create a properly configured EnhancedConfigRepository instance.

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 Key Vault integration	None
auto_load_to_environ	bool	If True, automatically load configs to os.environ on init	False

Returns:

Name	Type	Description
EnhancedConfigRepositoryInterface	EnhancedConfigRepositoryInterface	Configured repository instance

Usage

Simple usage with environment variables only

config_repo = create_enhanced_config_repository()

With App Config

config_repo = create_enhanced_config_repository( app_config_connection_string="Endpoint=...", auto_load_to_environ=True )

With App Config and Key Vault

secrets_repo = SecretsRepository(vault_url="...") config_repo = create_enhanced_config_repository( app_config_connection_string="Endpoint=...", secrets_repository=secrets_repo, auto_load_to_environ=True )

Source code in azure_bootstrap/repositories/enhanced_config_repository.py

def create_enhanced_config_repository(
    app_config_connection_string: str | None = None,
    secrets_repository: SecretsRepositoryInterface | None = None,
    auto_load_to_environ: bool = False,
) -> EnhancedConfigRepositoryInterface:
    """
    Factory function to create an enhanced configuration repository.

    This factory function provides a convenient way to create a properly
    configured EnhancedConfigRepository instance.

    Args:
        app_config_connection_string: Azure App Configuration connection string
        secrets_repository: Optional secrets repository for Key Vault integration
        auto_load_to_environ: If True, automatically load configs to os.environ on init

    Returns:
        EnhancedConfigRepositoryInterface: Configured repository instance

    Usage:
        # Simple usage with environment variables only
        config_repo = create_enhanced_config_repository()

        # With App Config
        config_repo = create_enhanced_config_repository(
            app_config_connection_string="Endpoint=...",
            auto_load_to_environ=True
        )

        # With App Config and Key Vault
        secrets_repo = SecretsRepository(vault_url="...")
        config_repo = create_enhanced_config_repository(
            app_config_connection_string="Endpoint=...",
            secrets_repository=secrets_repo,
            auto_load_to_environ=True
        )
    """
    logger.info(
        "Creating enhanced configuration repository",
        extra={
            "has_app_config": bool(app_config_connection_string),
            "has_secrets_repo": bool(secrets_repository),
            "auto_load": auto_load_to_environ,
            "operation": "create_repository",
        },
    )

    return EnhancedConfigRepository(
        app_config_connection_string=app_config_connection_string,
        secrets_repository=secrets_repository,
        auto_load_to_environ=auto_load_to_environ,
    )

compare_secrets

compare_secrets(a: str | bytes | None, b: str | bytes | None) -> bool

Constant-time equality. Returns False on any None / empty input.

Coerces str to bytes via UTF-8. Bytes inputs pass through unchanged.

Source code in azure_bootstrap/security/__init__.py

def compare_secrets(a: str | bytes | None, b: str | bytes | None) -> bool:
    """Constant-time equality. Returns False on any None / empty input.

    Coerces str to bytes via UTF-8. Bytes inputs pass through unchanged.
    """
    if not a or not b:
        return False
    a_b = a.encode("utf-8") if isinstance(a, str) else a
    b_b = b.encode("utf-8") if isinstance(b, str) else b
    return hmac.compare_digest(a_b, b_b)

initialize_application

initialize_application(secrets_repository: SecretsRepositoryInterface | None = None) -> EnhancedConfigRepositoryInterface

Convenience function to perform complete application bootstrap.

This is the main entry point for application initialization. It handles the complete bootstrap sequence and returns the configured repository.

Parameters:

Name	Type	Description	Default
secrets_repository	SecretsRepositoryInterface | None	Optional secrets repository for Key Vault integration	None

Returns:

Type	Description
EnhancedConfigRepositoryInterface	Configured enhanced configuration repository with all configs loaded

Example

Simple initialization

config_repo = initialize_application()

With secrets repository

from src.repositories.secrets_repository import create_secrets_repository secrets_repo = create_secrets_repository() config_repo = initialize_application(secrets_repository=secrets_repo)

After initialization, use standard os.environ access

app_insights_key = os.environ.get("APPLICATIONINSIGHTS_CONNECTION_STRING")

Source code in azure_bootstrap/services/application_bootstrap.py

def initialize_application(
    secrets_repository: SecretsRepositoryInterface | None = None,
) -> EnhancedConfigRepositoryInterface:
    """
    Convenience function to perform complete application bootstrap.

    This is the main entry point for application initialization. It handles
    the complete bootstrap sequence and returns the configured repository.

    Args:
        secrets_repository: Optional secrets repository for Key Vault integration

    Returns:
        Configured enhanced configuration repository with all configs loaded

    Example:
        # Simple initialization
        config_repo = initialize_application()

        # With secrets repository
        from src.repositories.secrets_repository import create_secrets_repository
        secrets_repo = create_secrets_repository()
        config_repo = initialize_application(secrets_repository=secrets_repo)

        # After initialization, use standard os.environ access
        app_insights_key = os.environ.get("APPLICATIONINSIGHTS_CONNECTION_STRING")
    """
    bootstrap = ApplicationBootstrap(secrets_repository=secrets_repository)
    repo = bootstrap.initialize()
    global _last_initialized_repo
    _last_initialized_repo = repo
    return repo

ensure_bootstrap_logging

ensure_bootstrap_logging() -> None

Ensure bootstrap logging is configured.

This is a convenience function that can be called safely multiple times. Call this early in your application startup before doing any logging.

Source code in azure_bootstrap/services/bootstrap_logging.py

def ensure_bootstrap_logging() -> None:
    """
    Ensure bootstrap logging is configured.

    This is a convenience function that can be called safely multiple times.
    Call this early in your application startup before doing any logging.
    """
    BootstrapLogger.configure_bootstrap_logging()

get_bootstrap_logger

get_bootstrap_logger(name: str) -> Logger

Get a logger that works during bootstrap phase.

Parameters:

Name	Type	Description	Default
name	str	Logger name (typically name)	required

Returns:

Type	Description
Logger	Logger configured for bootstrap use

Example

from src.infrastructure.bootstrap_logging import get_bootstrap_logger

logger = get_bootstrap_logger(name) logger.info("This works before full telemetry setup!")

Source code in azure_bootstrap/services/bootstrap_logging.py

def get_bootstrap_logger(name: str) -> logging.Logger:
    """
    Get a logger that works during bootstrap phase.

    Args:
        name: Logger name (typically __name__)

    Returns:
        Logger configured for bootstrap use

    Example:
        from src.infrastructure.bootstrap_logging import get_bootstrap_logger

        logger = get_bootstrap_logger(__name__)
        logger.info("This works before full telemetry setup!")
    """
    return BootstrapLogger.create_logger(name)

soft_fail

soft_fail(*, operation: str, catch: type[BaseException] | tuple[type[BaseException], ...] = Exception, alert_severity: str | None = 'error', counter_name: str | None = None, re_raise_unrecoverable: bool = True) -> Generator[dict[str, Any], None, None]

Context-manager form of :func:soft_fail_with.

Yields a mutable dict the caller can inspect after the block::

with soft_fail(operation='ai.summary', counter_name='ai.summary.failed') as ctx:
    summary = ai.summarize(text)
if ctx['degraded']:
    summary = None  # caller handles the degraded case

Source code in azure_bootstrap/softfail/__init__.py

@contextmanager
def soft_fail(
    *,
    operation: str,
    catch: type[BaseException] | tuple[type[BaseException], ...] = Exception,
    alert_severity: str | None = "error",
    counter_name: str | None = None,
    re_raise_unrecoverable: bool = True,
) -> Generator[dict[str, Any], None, None]:
    """Context-manager form of :func:`soft_fail_with`.

    Yields a mutable dict the caller can inspect after the block::

        with soft_fail(operation='ai.summary', counter_name='ai.summary.failed') as ctx:
            summary = ai.summarize(text)
        if ctx['degraded']:
            summary = None  # caller handles the degraded case
    """
    state: dict[str, Any] = {"degraded": False, "reason": None, "exception": None}
    try:
        yield state
    except catch as exc:
        if re_raise_unrecoverable and is_unrecoverable(exc):
            raise
        state["degraded"] = True
        state["reason"] = type(exc).__name__
        state["exception"] = exc
        _logger.warning(
            "soft-fail in %s: %s",
            operation,
            type(exc).__name__,
            exc_info=True,
            extra={
                "operation": operation,
                "exception_type": type(exc).__name__,
                "error": str(exc)[:500],
            },
        )
        if counter_name:
            bump_counter(counter_name)
        if alert_severity:
            _fire_alert(alert_severity, operation, exc)

soft_fail_with

soft_fail_with(fn: Callable[..., T], *args: Any, fallback: T, catch: type[BaseException] | tuple[type[BaseException], ...] = Exception, operation: str, alert_severity: str | None = 'error', counter_name: str | None = None, fallback_fn: Callable[[BaseException], T] | None = None, re_raise_unrecoverable: bool = True, **kwargs: Any) -> SoftFailResult[T]

Call fn(*args, **kwargs); soft-fail with fallback on caught error.

Unrecoverable exceptions (per :func:is_unrecoverable) propagate by default — set re_raise_unrecoverable=False to explicitly suppress them too.

Source code in azure_bootstrap/softfail/__init__.py

def soft_fail_with(
    fn: Callable[..., T],
    *args: Any,
    fallback: T,
    catch: type[BaseException] | tuple[type[BaseException], ...] = Exception,
    operation: str,
    alert_severity: str | None = "error",
    counter_name: str | None = None,
    fallback_fn: Callable[[BaseException], T] | None = None,
    re_raise_unrecoverable: bool = True,
    **kwargs: Any,
) -> SoftFailResult[T]:
    """Call ``fn(*args, **kwargs)``; soft-fail with ``fallback`` on caught error.

    Unrecoverable exceptions (per :func:`is_unrecoverable`) propagate by
    default — set ``re_raise_unrecoverable=False`` to explicitly suppress
    them too.
    """
    try:
        value = fn(*args, **kwargs)
    except catch as exc:
        if re_raise_unrecoverable and is_unrecoverable(exc):
            raise
        _logger.warning(
            "soft-fail in %s: %s",
            operation,
            type(exc).__name__,
            exc_info=True,
            extra={
                "operation": operation,
                "exception_type": type(exc).__name__,
                "error": str(exc)[:500],
            },
        )
        if counter_name:
            bump_counter(counter_name)
        if alert_severity:
            _fire_alert(alert_severity, operation, exc)
        resolved = fallback_fn(exc) if fallback_fn else fallback
        return SoftFailResult(
            value=resolved,
            degraded=True,
            reason=type(exc).__name__,
            exception=exc,
        )
    return SoftFailResult(value=value, degraded=False)

traced

traced(*, operation: str | None = None, alert_on_error: str | None = None, sensitive_args: tuple[str, ...] = (), log_result: bool = False, slow_threshold_seconds: float | None = None) -> Callable[[F], F]

Trace sync or async functions.

Records latency on every call (success + exception). Logs entry/exit at DEBUG only. When DEBUG is off, inspect.signature is skipped so the hot path stays cheap.

Source code in azure_bootstrap/tracing/decorators.py

def traced(
    *,
    operation: str | None = None,
    alert_on_error: str | None = None,
    sensitive_args: tuple[str, ...] = (),
    log_result: bool = False,
    slow_threshold_seconds: float | None = None,
) -> Callable[[F], F]:
    """Trace sync or async functions.

    Records latency on every call (success + exception). Logs entry/exit at
    DEBUG only. When DEBUG is off, ``inspect.signature`` is skipped so the
    hot path stays cheap.
    """

    def decorator(func: F) -> F:
        op = operation or f"{func.__module__}.{func.__qualname__}"
        is_async = asyncio.iscoroutinefunction(func)
        logger = logging.getLogger(func.__module__)

        if is_async:

            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                trace_id = uuid.uuid4().hex[:12]
                debug_on = logger.isEnabledFor(logging.DEBUG)
                masked_args: dict[str, str] | None = None
                if debug_on:
                    masked_args = _mask_args_for_log(func, args, kwargs, sensitive_args)
                    logger.debug("→ %s", op, extra=_build_entry_extra(op, trace_id, masked_args))
                start = time.monotonic()
                try:
                    result = await func(*args, **kwargs)
                except BaseException as exc:
                    elapsed = round(time.monotonic() - start, 3)
                    _record_latency(op, elapsed, error=True, slow=False)
                    logger.exception(
                        "✗ %s raised %s after %ss",
                        op,
                        type(exc).__name__,
                        elapsed,
                        extra={
                            "operation": op,
                            "trace_id": trace_id,
                            "elapsed_seconds": elapsed,
                            "exception_type": type(exc).__name__,
                        },
                    )
                    _maybe_alert_error(op, exc, alert_on_error, logger)
                    raise
                elapsed = round(time.monotonic() - start, 3)
                threshold = _resolve_threshold(op, slow_threshold_seconds)
                is_slow = threshold is not None and elapsed > threshold
                _record_latency(op, elapsed, error=False, slow=is_slow)
                if debug_on:
                    exit_extra: dict[str, Any] = {
                        "operation": op,
                        "trace_id": trace_id,
                        "elapsed_seconds": elapsed,
                    }
                    if log_result:
                        exit_extra["result"] = _safe_repr(result)
                    logger.debug("✓ %s ok in %.3fs", op, elapsed, extra=exit_extra)
                if is_slow and threshold is not None:
                    _slow_alert(op, elapsed, threshold)
                return result

            return cast(F, async_wrapper)

        @functools.wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            trace_id = uuid.uuid4().hex[:12]
            debug_on = logger.isEnabledFor(logging.DEBUG)
            masked_args: dict[str, str] | None = None
            if debug_on:
                masked_args = _mask_args_for_log(func, args, kwargs, sensitive_args)
                logger.debug("→ %s", op, extra=_build_entry_extra(op, trace_id, masked_args))
            start = time.monotonic()
            try:
                result = func(*args, **kwargs)
            except BaseException as exc:
                elapsed = round(time.monotonic() - start, 3)
                _record_latency(op, elapsed, error=True, slow=False)
                logger.exception(
                    "✗ %s raised %s after %ss",
                    op,
                    type(exc).__name__,
                    elapsed,
                    extra={
                        "operation": op,
                        "trace_id": trace_id,
                        "elapsed_seconds": elapsed,
                        "exception_type": type(exc).__name__,
                    },
                )
                _maybe_alert_error(op, exc, alert_on_error, logger)
                raise
            elapsed = round(time.monotonic() - start, 3)
            threshold = _resolve_threshold(op, slow_threshold_seconds)
            is_slow = threshold is not None and elapsed > threshold
            _record_latency(op, elapsed, error=False, slow=is_slow)
            if debug_on:
                exit_extra = {
                    "operation": op,
                    "trace_id": trace_id,
                    "elapsed_seconds": elapsed,
                }
                if log_result:
                    exit_extra["result"] = _safe_repr(result)
                logger.debug("✓ %s ok in %.3fs", op, elapsed, extra=exit_extra)
            if is_slow and threshold is not None:
                _slow_alert(op, elapsed, threshold)
            return result

        return cast(F, sync_wrapper)

    return decorator

configure_transports

configure_transports(*, console: bool | None = None, app_insights: bool | None = None, sumo_logic: bool | None = None, panther: bool | None = None, file: bool | None = None, blob: bool | None = None, sql: bool | None = None, nosql: bool | None = None, adx: bool | None = None, event_hubs: bool | None = None) -> None

Enable/disable the ten built-in transports. Idempotent and re-runnable.

For each sink, an explicit boolean wins; otherwise the per-transport env flag is consulted (defaults: console on, all others off).

============== ============================== parameter env flag ============== ============================== console CONSOLE_LOGGING_ENABLED app_insights APP_INSIGHTS_LOGGING_ENABLED sumo_logic SUMO_LOGIC_LOGGING_ENABLED panther PANTHER_LOGGING_ENABLED file FILE_LOGGING_ENABLED blob BLOB_LOGGING_ENABLED sql SQL_LOGGING_ENABLED nosql NOSQL_LOGGING_ENABLED ============== ==============================

Also sets the root logger to effective_log_level() so enabled transports actually receive records (the stdlib root default of WARNING would otherwise drop INFO/DEBUG). Honors the LOG_LEVEL / DEBUG_LOGGING_ENABLED env contract shared with configure_logging().

Source code in azure_bootstrap/transports/__init__.py

def configure_transports(
    *,
    console: bool | None = None,
    app_insights: bool | None = None,
    sumo_logic: bool | None = None,
    panther: bool | None = None,
    file: bool | None = None,
    blob: bool | None = None,
    sql: bool | None = None,
    nosql: bool | None = None,
    adx: bool | None = None,
    event_hubs: bool | None = None,
) -> None:
    """Enable/disable the ten built-in transports. Idempotent and re-runnable.

    For each sink, an explicit boolean wins; otherwise the per-transport env flag
    is consulted (defaults: console on, all others off).

    ==============  ==============================
    parameter       env flag
    ==============  ==============================
    console         CONSOLE_LOGGING_ENABLED
    app_insights    APP_INSIGHTS_LOGGING_ENABLED
    sumo_logic      SUMO_LOGIC_LOGGING_ENABLED
    panther         PANTHER_LOGGING_ENABLED
    file            FILE_LOGGING_ENABLED
    blob            BLOB_LOGGING_ENABLED
    sql             SQL_LOGGING_ENABLED
    nosql           NOSQL_LOGGING_ENABLED
    ==============  ==============================

    Also sets the root logger to ``effective_log_level()`` so enabled transports
    actually receive records (the stdlib root default of WARNING would otherwise
    drop INFO/DEBUG). Honors the ``LOG_LEVEL`` / ``DEBUG_LOGGING_ENABLED`` env
    contract shared with ``configure_logging()``.
    """
    logging.getLogger().setLevel(effective_log_level())
    params = {
        "console": console,
        "app_insights": app_insights,
        "sumo_logic": sumo_logic,
        "panther": panther,
        "file": file,
        "blob": blob,
        "sql": sql,
        "nosql": nosql,
        "adx": adx,
        "event_hubs": event_hubs,
    }
    for name, param in params.items():
        if _resolve(param, _ENV_FLAGS[name], _DEFAULTS[name]):
            enable_transport(name)
        else:
            disable_transport(name)

disable_transport

disable_transport(name: str) -> bool

Detach (and close) the transport's handler. Idempotent.

Returns True if a handler was removed, False if it was not enabled.

Source code in azure_bootstrap/transports/__init__.py

def disable_transport(name: str) -> bool:
    """Detach (and close) the transport's handler. Idempotent.

    Returns True if a handler was removed, False if it was not enabled.
    """
    with _lock:
        handler = _active.pop(name, None)
        if handler is None:
            return False
        try:
            logging.getLogger().removeHandler(handler)
        finally:
            try:
                handler.close()
            except Exception:
                pass
        return True

enable_transport

enable_transport(name: str) -> bool

Attach the transport's handler to the root logger. Idempotent.

Returns True if a handler was added, False if the transport was already enabled or its factory returned None (transport unavailable).

Source code in azure_bootstrap/transports/__init__.py

def enable_transport(name: str) -> bool:
    """Attach the transport's handler to the root logger. Idempotent.

    Returns True if a handler was added, False if the transport was already
    enabled or its factory returned ``None`` (transport unavailable).
    """
    with _lock:
        _reconcile()
        factory = _factories.get(name)
        if factory is None:
            raise ValueError(f"transport {name!r} is not registered")
        if name in _active:
            return False
        try:
            handler = factory()
        except Exception:
            logging.getLogger(__name__).debug("transport %r factory raised", name, exc_info=True)
            return False
        if handler is None:
            return False
        handler._ab_transport = name  # type: ignore[attr-defined]
        logging.getLogger().addHandler(handler)
        _active[name] = handler
        return True

list_transports

list_transports() -> dict[str, dict[str, Any]]

Return {name: {"registered": True, "enabled": bool}} for all transports.

Source code in azure_bootstrap/transports/__init__.py

def list_transports() -> dict[str, dict[str, Any]]:
    """Return ``{name: {"registered": True, "enabled": bool}}`` for all transports."""
    with _lock:
        _reconcile()
        return {name: {"registered": True, "enabled": name in _active} for name in _factories}

register_transport

register_transport(name: str, factory: TransportFactory, *, replace: bool = False) -> None

Register a transport factory under name.

Raises ValueError if name is already registered and replace is False. Re-registering a currently-enabled transport disables it first; the caller must re-enable to pick up the new factory.

Source code in azure_bootstrap/transports/__init__.py

def register_transport(name: str, factory: TransportFactory, *, replace: bool = False) -> None:
    """Register a transport factory under ``name``.

    Raises ``ValueError`` if ``name`` is already registered and ``replace`` is
    False. Re-registering a currently-enabled transport disables it first; the
    caller must re-enable to pick up the new factory.
    """
    if not isinstance(name, str) or not name:
        raise ValueError("transport name must be a non-empty string")
    with _lock:
        if name in _factories and not replace:
            raise ValueError(f"transport {name!r} is already registered (pass replace=True)")
        if name in _active:
            disable_transport(name)
        _factories[name] = factory

queue_message_schema

queue_message_schema(*, required_fields: Iterable[str] = ('correlation_id',), path_field: str | None = None, path_required_prefix: str | None = None, counter_namespace: str = 'queue_message') -> MessageSchema

Build a sensible MessageSchema for the common consumer case.

Adds path-traversal defense (forbidden substrings .. and ://) when path_field is supplied. required_prefix enforces a blob / container scoping convention.

Source code in azure_bootstrap/validation/__init__.py

def queue_message_schema(
    *,
    required_fields: Iterable[str] = ("correlation_id",),
    path_field: str | None = None,
    path_required_prefix: str | None = None,
    counter_namespace: str = "queue_message",
) -> MessageSchema:
    """Build a sensible MessageSchema for the common consumer case.

    Adds path-traversal defense (forbidden substrings ``..`` and ``://``)
    when ``path_field`` is supplied. ``required_prefix`` enforces a blob /
    container scoping convention.
    """
    rules: list[FieldRule] = [
        FieldRule(name=name, required=True, type=str, non_empty=True) for name in required_fields
    ]
    if path_field is not None:
        rules.append(
            FieldRule(
                name=path_field,
                required=True,
                type=str,
                non_empty=True,
                forbidden_substrings=("..", "://"),
                required_prefix=path_required_prefix,
            )
        )
    return MessageSchema(fields=tuple(rules), counter_namespace=counter_namespace)

validate_message

validate_message(payload: Any, schema: MessageSchema, *, raise_unrecoverable: bool = True) -> dict[str, Any]

Validate payload against schema. Returns the dict on success.

Failures (non-dict, missing field, type mismatch, pattern, forbidden substring/prefix, required_prefix) bump the namespaced counter and raise :class:InvalidMessageError by default.

Source code in azure_bootstrap/validation/__init__.py

def validate_message(
    payload: Any,
    schema: MessageSchema,
    *,
    raise_unrecoverable: bool = True,
) -> dict[str, Any]:
    """Validate ``payload`` against ``schema``. Returns the dict on success.

    Failures (non-dict, missing field, type mismatch, pattern, forbidden
    substring/prefix, required_prefix) bump the namespaced counter and
    raise :class:`InvalidMessageError` by default.
    """
    if not isinstance(payload, dict):
        _bump_rejection(schema)
        if raise_unrecoverable:
            raise InvalidMessageError(
                f"payload is not a JSON object (got {type(payload).__name__})"
            )
        return {}

    for rule in schema.fields:
        reason = _check_field(rule, payload)
        if reason:
            _bump_rejection(schema)
            _logger.warning(
                "validate_message: rejected — %s",
                reason,
                extra={
                    "operation": "validate_message",
                    "schema_namespace": schema.counter_namespace,
                    "reason": reason,
                },
            )
            if raise_unrecoverable:
                raise InvalidMessageError(reason)
            return {}

    return payload

refresh_setting

refresh_setting(*names: str) -> None

Re-read named settings from the cached App Configuration repo and write their values into os.environ.

Net-new in v2. Designed to be called from a recurring job (see azure_bootstrap.config_refresh.refresh_log_flags) so ops can flip a setting in App Configuration and see it take effect within seconds without redeploying.

No-ops with a DEBUG log when initialize_application() has not yet run. Best-effort — never raises.

Source code in azure_bootstrap/__init__.py

def refresh_setting(*names: str) -> None:
    """Re-read named settings from the cached App Configuration repo and
    write their values into ``os.environ``.

    Net-new in v2. Designed to be called from a recurring job (see
    ``azure_bootstrap.config_refresh.refresh_log_flags``) so ops can flip a
    setting in App Configuration and see it take effect within seconds
    without redeploying.

    No-ops with a DEBUG log when ``initialize_application()`` has not yet
    run. Best-effort — never raises.
    """
    if not names:
        return
    logger = _stdlib_logging.getLogger(__name__)
    try:
        from azure_bootstrap.services.application_bootstrap import (
            get_last_initialized_repo,
        )
    except Exception:
        logger.debug("refresh_setting: bootstrap module unavailable")
        return
    repo = get_last_initialized_repo()
    if repo is None:
        logger.debug("refresh_setting: no cached repo (initialize_application not called)")
        return
    for name in names:
        if not isinstance(name, str) or not name:
            continue
        try:
            value = repo.get_value(name)
        except Exception as exc:
            logger.warning("refresh_setting: failed to read %s: %s", name, exc)
            continue
        if value is None:
            continue
        _os.environ[name] = str(value)