Configuration Management, Secrets Lifecycle, and Multi-Environment Drift Control¶

# config/static_config.py
from pydantic import BaseSettings, Field
from typing import Literal

class DatabaseConfig(BaseSettings):
    """Database connection configuration"""
    host: str = Field(..., description="Database host")
    port: int = Field(5432, description="Database port")
    database: str = Field(..., description="Database name")
    pool_size: int = Field(10, ge=1, le=100, description="Connection pool size")
    pool_timeout: int = Field(30, ge=1, description="Pool timeout in seconds")

    class Config:
        env_prefix = "DB_"
        case_sensitive = False

class AppConfig(BaseSettings):
    """Application configuration"""
    environment: Literal["dev", "stage", "prod"] = Field(..., description="Environment")
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = Field("INFO")
    api_timeout: int = Field(30, ge=1, description="API timeout in seconds")

    database: DatabaseConfig = Field(..., description="Database config")

    class Config:
        env_prefix = "APP_"
        case_sensitive = False

# Usage
config = AppConfig()

# config/feature_flags.py
import redis
import json

class FeatureFlags:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def is_enabled(self, flag: str, user_id: str = None) -> bool:
        """Check if feature flag is enabled"""
        key = f"feature_flag:{flag}"

        # Check global flag
        global_value = self.redis.get(key)
        if global_value:
            config = json.loads(global_value)
            if config.get("enabled", False):
                return True

        # Check user-specific flag
        if user_id:
            user_key = f"feature_flag:{flag}:{user_id}"
            user_value = self.redis.get(user_key)
            if user_value:
                return json.loads(user_value).get("enabled", False)

        return False

    def set_flag(self, flag: str, enabled: bool, user_id: str = None):
        """Set feature flag"""
        key = f"feature_flag:{flag}" if not user_id else f"feature_flag:{flag}:{user_id}"
        self.redis.set(
            key,
            json.dumps({"enabled": enabled, "updated_at": datetime.now().isoformat()}),
            ex=86400  # 24 hour TTL
        )

# config/ml_config.py
from pydantic import BaseSettings, SecretStr

class MLConfig(BaseSettings):
    """ML model configuration"""
    model_endpoint: str = Field(..., description="ONNX model endpoint")
    feature_store_endpoint: str = Field(..., description="Feature store endpoint")
    inference_timeout: int = Field(5, ge=1, description="Inference timeout")
    batch_size: int = Field(32, ge=1, le=128, description="Batch size")

    # Sensitive - use SecretStr
    api_key: SecretStr = Field(..., description="ML API key")

    class Config:
        env_prefix = "ML_"

# config/gis_config.py
class GISConfig(BaseSettings):
    """Geospatial configuration"""
    tile_endpoint: str = Field(..., description="Vector tile endpoint")
    basemap_url: str = Field(..., description="Basemap URL")
    max_zoom: int = Field(12, ge=1, le=18, description="Maximum zoom level")
    cache_ttl: int = Field(3600, ge=0, description="Cache TTL in seconds")

    # Sensitive bucket paths
    tile_bucket: str = Field(..., description="Tile storage bucket")
    data_bucket: str = Field(..., description="Geospatial data bucket")

    class Config:
        env_prefix = "GIS_"

# config/schema.py
from pydantic import BaseModel, validator
from typing import Literal

class ConfigSchema(BaseModel):
    environment: Literal["dev", "stage", "prod"]
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"]
    api_timeout: int

    @validator('api_timeout')
    def validate_timeout(cls, v):
        if v < 1 or v > 300:
            raise ValueError('api_timeout must be between 1 and 300')
        return v

# scripts/promote_config.py
import subprocess
import sys
from pathlib import Path

def promote_config(source_env: str, target_env: str, config_path: Path):
    """Promote configuration from source to target environment"""
    # 1. Validate source config
    print(f"Validating {source_env} configuration...")
    if not validate_config(config_path / f"values-{source_env}.yaml"):
        print(f"ERROR: {source_env} config validation failed")
        sys.exit(1)

    # 2. Check for drift
    print(f"Checking for drift in {target_env}...")
    drift = check_drift(source_env, target_env)
    if drift:
        print(f"WARNING: Drift detected between {source_env} and {target_env}")
        if not confirm("Continue with promotion?"):
            sys.exit(1)

    # 3. Create promotion PR
    print(f"Creating promotion PR from {source_env} to {target_env}...")
    pr_url = create_promotion_pr(source_env, target_env, config_path)
    print(f"Promotion PR: {pr_url}")

    # 4. Run validation in target environment
    print(f"Running validation in {target_env}...")
    if not validate_in_target(target_env):
        print(f"ERROR: Validation failed in {target_env}")
        sys.exit(1)

    # 5. Apply config (if auto-approve)
    if target_env != "prod":
        print(f"Applying configuration to {target_env}...")
        apply_config(target_env, config_path)
    else:
        print("Production promotion requires manual approval")
        print(f"Review PR: {pr_url}")

def validate_config(config_file: Path) -> bool:
    """Validate configuration file"""
    result = subprocess.run(
        ["helm", "lint", "-f", str(config_file)],
        capture_output=True
    )
    return result.returncode == 0

def check_drift(source_env: str, target_env: str) -> bool:
    """Check for configuration drift"""
    # Compare configs
    source_config = load_config(f"values-{source_env}.yaml")
    target_config = load_config(f"values-{target_env}.yaml")

    return source_config != target_config

# scripts/detect_postgres_drift.py
import psycopg2
from schema_diff import compare_schemas

def detect_postgres_drift(dev_conn, prod_conn):
    """Detect schema drift between dev and prod"""
    dev_schema = get_schema(dev_conn)
    prod_schema = get_schema(prod_conn)

    diff = compare_schemas(dev_schema, prod_schema)

    if diff:
        print("Schema drift detected:")
        for change in diff:
            print(f"  - {change}")
        return True

    return False

def get_schema(conn):
    """Get database schema"""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT 
                schemaname,
                tablename,
                columnname,
                datatype
            FROM pg_catalog.pg_table_def
            WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
            ORDER BY schemaname, tablename, columnname
        """)
        return cur.fetchall()

# scripts/detect_redis_drift.py
import redis

def detect_redis_drift(dev_client: redis.Redis, prod_client: redis.Redis):
    """Detect Redis configuration drift"""
    dev_config = dev_client.config_get()
    prod_config = prod_client.config_get()

    drift = []
    for key in set(dev_config.keys()) | set(prod_config.keys()):
        if dev_config.get(key) != prod_config.get(key):
            drift.append({
                "key": key,
                "dev": dev_config.get(key),
                "prod": prod_config.get(key)
            })

    return drift

# scripts/create_secret.py
from cryptography.fernet import Fernet
import secrets
import json

class SecretManager:
    def __init__(self, vault_client):
        self.vault = vault_client

    def create_secret(
        self,
        secret_name: str,
        secret_type: str,
        environment: str,
        scopes: list,
        expires_days: int = 90
    ) -> dict:
        """Create new secret with lifecycle policy"""
        # Generate secret value
        if secret_type == "api_key":
            value = secrets.token_urlsafe(32)
        elif secret_type == "password":
            value = secrets.token_urlsafe(24)
        elif secret_type == "jwt_signing_key":
            value = Fernet.generate_key().decode()
        else:
            raise ValueError(f"Unknown secret type: {secret_type}")

        # Store in Vault
        self.vault.write(
            f"secret/{environment}/{secret_name}",
            value=value,
            type=secret_type,
            scopes=scopes,
            created_at=datetime.now().isoformat(),
            expires_at=(datetime.now() + timedelta(days=expires_days)).isoformat()
        )

        return {
            "name": secret_name,
            "value": value,  # Only returned once
            "expires_at": (datetime.now() + timedelta(days=expires_days)).isoformat()
        }

# scripts/rotate_db_password.py
import asyncpg
from vault_client import VaultClient

class DatabasePasswordRotator:
    def __init__(self, vault: VaultClient, db_pool: asyncpg.Pool):
        self.vault = vault
        self.db_pool = db_pool

    async def rotate_password(self, user: str, environment: str):
        """Rotate database password with zero downtime"""
        # 1. Generate new password
        new_password = secrets.token_urlsafe(24)

        # 2. Create new user with new password (dual-secret pattern)
        new_user = f"{user}_new"
        async with self.db_pool.acquire() as conn:
            await conn.execute(f"""
                CREATE USER {new_user} WITH PASSWORD %s;
                GRANT ALL PRIVILEGES ON DATABASE mydb TO {new_user};
            """, new_password)

        # 3. Update application config (grace period)
        await self.update_app_config(new_user, new_password, grace_period=True)

        # 4. Wait for connections to migrate
        await asyncio.sleep(300)  # 5 minute grace period

        # 5. Drop old user
        async with self.db_pool.acquire() as conn:
            await conn.execute(f"DROP USER {user}")

        # 6. Rename new user
        async with self.db_pool.acquire() as conn:
            await conn.execute(f"ALTER USER {new_user} RENAME TO {user}")

        # 7. Update Vault
        self.vault.write(
            f"secret/{environment}/database/{user}",
            password=new_password,
            rotated_at=datetime.now().isoformat()
        )

# scripts/retire_secret.py
class SecretRetirer:
    def __init__(self, vault: VaultClient):
        self.vault = vault

    def retire_secret(self, secret_name: str, environment: str, reason: str):
        """Retire and archive secret"""
        # 1. Mark as retired
        secret_data = self.vault.read(f"secret/{environment}/{secret_name}")
        secret_data["retired"] = True
        secret_data["retired_at"] = datetime.now().isoformat()
        secret_data["retirement_reason"] = reason

        # 2. Move to archive
        self.vault.write(
            f"archive/{environment}/{secret_name}",
            **secret_data
        )

        # 3. Delete from active secrets
        self.vault.delete(f"secret/{environment}/{secret_name}")

        # 4. Revoke all tokens using this secret
        self.revoke_tokens(secret_name, environment)

# config/validator.py
from pydantic import BaseModel, ValidationError
import yaml

def validate_config_file(config_path: Path, schema: BaseModel):
    """Validate configuration file against schema"""
    with open(config_path) as f:
        config_data = yaml.safe_load(f)

    try:
        validated = schema(**config_data)
        return validated
    except ValidationError as e:
        print(f"Validation errors in {config_path}:")
        for error in e.errors():
            print(f"  - {error['loc']}: {error['msg']}")
        raise

# nicegui/config.py
from nicegui import app
from pydantic import BaseSettings
import os

class NiceGUIConfig(BaseSettings):
    """NiceGUI application configuration"""
    title: str = "My App"
    dark_mode: bool = True
    api_endpoint: str
    database_url: str
    redis_url: str

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# Load config
config = NiceGUIConfig()

# Use in app
app.config = config

# fastapi/config.py
from fastapi import FastAPI
from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str
    redis_url: str
    api_key: str

    class Config:
        env_file = ".env"

settings = Settings()

app = FastAPI(title=settings.app_name)

@app.get("/health")
async def health():
    return {"status": "healthy", "config_loaded": True}

# mlflow/config.py
import mlflow
from pydantic import BaseSettings

class MLflowConfig(BaseSettings):
    tracking_uri: str
    experiment_name: str = "default"
    registry_uri: str = None

    class Config:
        env_prefix = "MLFLOW_"

config = MLflowConfig()

# Configure MLflow
mlflow.set_tracking_uri(config.tracking_uri)
mlflow.set_experiment(config.experiment_name)

# tests/test_config.py
import pytest
from config import AppConfig, DatabaseConfig

def test_config_loading():
    """Test configuration loading"""
    config = AppConfig(
        environment="dev",
        log_level="INFO",
        database=DatabaseConfig(
            host="localhost",
            port=5432,
            database="testdb"
        )
    )

    assert config.environment == "dev"
    assert config.database.host == "localhost"

def test_config_validation():
    """Test configuration validation"""
    with pytest.raises(ValueError):
        AppConfig(
            environment="invalid",
            log_level="INFO"
        )

# tests/test_config_integration.py
@pytest.mark.integration
def test_config_deployment():
    """Test configuration deployment"""
    # Deploy config to test cluster
    subprocess.run(["kubectl", "apply", "-f", "k8s/test/"])

    # Verify config loaded
    result = subprocess.run(
        ["kubectl", "get", "configmap", "app-config"],
        capture_output=True
    )
    assert result.returncode == 0

# monitoring/config_metrics.py
from prometheus_client import Counter, Gauge, Histogram

config_load_errors = Counter(
    'config_load_errors_total',
    'Total config load errors',
    ['environment', 'config_type']
)

config_drift_detected = Gauge(
    'config_drift_detected',
    'Configuration drift detected',
    ['environment']
)

secret_expiration_days = Gauge(
    'secret_expiration_days',
    'Days until secret expiration',
    ['secret_name', 'environment']
)