🐍 第32课：配置管理

—— 动态配置+环境变量：安全灵活的配置体系

📌 本课目标

1️⃣ 多层配置体系

"""配置优先级（从低到高）：
1. 代码默认值     → 最基础
2. 配置文件       → 项目级
3. 环境变量       → 部署级
4. 命令行参数     → 运行时覆盖

原则：越外层优先级越高，环境变量 > 配置文件 > 默认值
"""

import os
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class DatabaseConfig:
    host: str = "localhost"
    port: int = 5432
    name: str = "myapp"
    user: str = "root"
    password: str = ""

@dataclass
class AppConfig:
    app_name: str = "MyApp"
    debug: bool = False
    log_level: str = "INFO"
    database: DatabaseConfig = field(default_factory=DatabaseConfig)
    
    @classmethod
    def from_env(cls):
        """从环境变量加载配置"""
        return cls(
            app_name=os.getenv("APP_NAME", "MyApp"),
            debug=os.getenv("DEBUG", "false").lower() in ("true", "1", "yes"),
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            database=DatabaseConfig(
                host=os.getenv("DB_HOST", "localhost"),
                port=int(os.getenv("DB_PORT", "5432")),
                name=os.getenv("DB_NAME", "myapp"),
                user=os.getenv("DB_USER", "root"),
                password=os.getenv("DB_PASSWORD", ""),
            ),
        )

# 使用
config = AppConfig.from_env()
print(f"应用: {config.app_name}")
print(f"调试: {config.debug}")
print(f"数据库: {config.database.host}:{config.database.port}")

2️⃣ .env 文件管理

"""python-dotenv 从 .env 文件加载环境变量

.env 文件示例：
APP_NAME=ProductionApp
DEBUG=false
LOG_LEVEL=WARNING
DB_HOST=db.example.com
DB_PORT=5432
DB_NAME=prod_db
DB_USER=app_user
DB_PASSWORD=super_secret_password
"""

# 安装：pip install python-dotenv

# 方式一：手动加载
from dotenv import load_dotenv
load_dotenv()  # 加载 .env 文件
import os
db_host = os.getenv("DB_HOST")

# 方式二：指定文件
load_dotenv("/path/to/.env.production")

# 方式三：dotenv_values（不修改 os.environ）
from dotenv import dotenv_values
config = dotenv_values(".env")
print(config)

# 方式四：查找 .env
from pathlib import Path
env_path = Path(__file__).parent / ".env"
load_dotenv(env_path)

3️⃣ 配置文件管理器

import json
import os
from pathlib import Path
from typing import Any, Dict

class ConfigManager:
    """多层配置管理器"""
    
    def __init__(self, config_dir="config"):
        self.config_dir = Path(config_dir)
        self._config: Dict[str, Any] = {}
        self._load_all()
    
    def _load_all(self):
        """加载所有配置层"""
        # 1. 默认配置
        defaults = self._load_json(self.config_dir / "default.json")
        self._config = defaults.copy()
        
        # 2. 环境配置（覆盖默认值）
        env_name = os.getenv("APP_ENV", "development")
        env_config = self._load_json(self.config_dir / f"{env_name}.json")
        self._deep_update(self._config, env_config)
        
        # 3. 环境变量覆盖
        self._override_from_env()
    
    def _load_json(self, path):
        """加载 JSON 配置文件"""
        if path.exists():
            with open(path) as f:
                return json.load(f)
        return {}
    
    def _deep_update(self, base, override):
        """深层合并字典"""
        for key, value in override.items():
            if key in base and isinstance(base[key], dict) and isinstance(value, dict):
                self._deep_update(base[key], value)
            else:
                base[key] = value
    
    def _override_from_env(self):
        """环境变量覆盖（APP_ 前缀）"""
        for key, value in os.environ.items():
            if key.startswith("APP_"):
                config_key = key[4:].lower()
                self._config[config_key] = value
    
    def get(self, key, default=None):
        """获取配置值（支持点号路径）"""
        keys = key.split(".")
        value = self._config
        for k in keys:
            if isinstance(value, dict):
                value = value.get(k)
            else:
                return default
            if value is None:
                return default
        return value
    
    def set(self, key, value):
        """设置配置值"""
        keys = key.split(".")
        config = self._config
        for k in keys[:-1]:
            config = config.setdefault(k, {})
        config[keys[-1]] = value
    
    @property
    def all(self):
        return self._config.copy()

# 使用示例（需要配置文件存在时）
# mgr = ConfigManager()
# db_host = mgr.get("database.host", "localhost")
# mgr.set("app.debug", True)

4️⃣ 敏感信息保护

"""敏感信息保护最佳实践：
1. 密码/密钥绝不入 Git → .gitignore 添加 .env
2. 使用环境变量传递 → 不硬编码
3. 生产环境用密钥管理服务 → AWS Secrets Manager / Vault
4. 日志脱敏 → 不输出密码字段
"""

import re
import os

def mask_sensitive(data, keys=None):
    """对敏感字段脱敏"""
    if keys is None:
        keys = ["password", "secret", "token", "key", "api_key"]
    
    if isinstance(data, dict):
        return {
            k: ("***" if any(s in k.lower() for s in keys) else mask_sensitive(v, keys))
            for k, v in data.items()
        }
    elif isinstance(data, str):
        # 掩盖手机号
        data = re.sub(r"(1[3-9]\d)\d{4}(\d{4})", r"\1****\2", data)
        return data
    return data

# 日志脱敏
def safe_log(data):
    """安全日志输出"""
    import json
    masked = mask_sensitive(data)
    return json.dumps(masked, ensure_ascii=False)

# 示例
config = {
    "app_name": "MyApp",
    "database": {
        "host": "localhost",
        "password": "super_secret",
        "api_key": "sk-123456",
    }
}
print(f"安全日志: {safe_log(config)}")
# 安全日志: {"app_name": "MyApp", "database": {"host": "localhost", "password": "***", "api_key": "***"}}

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第32课 配置管理验证"""
import os
import json
import tempfile

def test_dataclass_config():
    from dataclasses import dataclass
    
    @dataclass
    class Config:
        name: str = "default"
        debug: bool = False
        port: int = 8000
    
    c = Config()
    assert c.name == "default"
    assert c.debug == False
    assert c.port == 8000
    
    c2 = Config(name="prod", debug=True, port=80)
    assert c2.name == "prod"
    print("✅ dataclass配置测试通过")

def test_env_override():
    os.environ["TEST_VAR"] = "from_env"
    value = os.getenv("TEST_VAR", "default")
    assert value == "from_env"
    del os.environ["TEST_VAR"]
    print("✅ 环境变量覆盖测试通过")

def test_json_config():
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = os.path.join(tmpdir, "config.json")
        config = {"host": "localhost", "port": 5432, "debug": False}
        with open(config_path, "w") as f:
            json.dump(config, f)
        
        with open(config_path) as f:
            loaded = json.load(f)
        
        assert loaded["host"] == "localhost"
        assert loaded["port"] == 5432
    print("✅ JSON配置测试通过")

def test_mask_sensitive():
    import re
    
    def mask_sensitive(data, keys=None):
        if keys is None:
            keys = ["password", "secret", "token", "key"]
        if isinstance(data, dict):
            return {
                k: ("***" if any(s in k.lower() for s in keys) else mask_sensitive(v, keys))
                for k, v in data.items()
            }
        return data
    
    data = {
        "username": "admin",
        "password": "secret123",
        "api_key": "sk-xxx",
        "normal_field": "ok",
    }
    masked = mask_sensitive(data)
    assert masked["username"] == "admin"
    assert masked["password"] == "***"
    assert masked["api_key"] == "***"
    assert masked["normal_field"] == "ok"
    print("✅ 敏感信息脱敏测试通过")

if __name__ == "__main__":
    test_dataclass_config()
    test_env_override()
    test_json_config()
    test_mask_sensitive()
    print("\n🎉 第32课全部验证通过！")

6️⃣ 配置热更新

import json
import hashlib
from pathlib import Path
from typing import Callable

class ConfigWatcher:
    """配置文件监听与热更新"""
    
    def __init__(self, config_path, on_change: Callable):
        self.config_path = Path(config_path)
        self.on_change = on_change
        self._last_hash = None
    
    def check_and_reload(self):
        current_hash = self._get_file_hash()
        if current_hash != self._last_hash:
            self._last_hash = current_hash
            config = self._load_config()
            self.on_change(config)
            return True
        return False
    
    def _get_file_hash(self):
        if not self.config_path.exists():
            return None
        return hashlib.md5(self.config_path.read_bytes()).hexdigest()
    
    def _load_config(self):
        if not self.config_path.exists():
            return {}
        with open(self.config_path) as f:
            return json.load(f)

# 使用
def on_config_change(new_config):
    print(f"配置已更新: {new_config}")

# watcher = ConfigWatcher("config/app.json", on_config_change)

7️⃣ 配置验证与转换

from dataclasses import dataclass, field
from typing import Optional, List
import os

@dataclass
class ServerConfig:
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 4
    debug: bool = False
    
    def __post_init__(self):
        if not (1 <= self.port <= 65535):
            raise ValueError(f"端口必须在 1-65535 之间")
        if self.workers < 1:
            raise ValueError(f"workers 必须 >= 1")

@dataclass
class AppConfig:
    app_name: str = "MyApp"
    server: ServerConfig = field(default_factory=ServerConfig)
    allowed_hosts: List[str] = field(default_factory=lambda: ["*"])
    
    @classmethod
    def from_env(cls):
        return cls(
            app_name=os.getenv("APP_NAME", "MyApp"),
            server=ServerConfig(
                host=os.getenv("SERVER_HOST", "0.0.0.0"),
                port=int(os.getenv("SERVER_PORT", "8000")),
            ),
        )

# 使用
config = AppConfig.from_env()
print(f"应用: {config.app_name} @ {config.server.host}:{config.server.port}")

8️⃣ 12-Factor App 配置原则

"""12-Factor App 关于配置的核心原则：
- 严格分离配置和代码
- 配置存储在环境变量中
- 不在代码中硬编码环境差异

环境差异示例：
- 开发：localhost, debug=true, 本地数据库
- 测试：test-db, debug=false
- 生产：prod-db, debug=false, 密钥管理服务
"""

import os
from dataclasses import dataclass, field
from typing import List

@dataclass
class FullAppConfig:
    """完整应用配置（12-Factor 风格）"""
    
    # 环境
    env: str = "development"
    debug: bool = True
    
    # 服务器
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 1
    
    # 数据库
    db_url: str = "sqlite:///dev.db"
    db_pool_size: int = 5
    
    # Redis
    redis_url: str = "redis://localhost:6379/0"
    
    # 安全
    secret_key: str = "dev-secret-key-change-in-prod"
    allowed_hosts: List[str] = field(default_factory=lambda: ["*"])
    
    # 日志
    log_level: str = "DEBUG"
    log_file: str = ""
    
    @classmethod
    def from_env(cls):
        return cls(
            env=os.getenv("APP_ENV", "development"),
            debug=os.getenv("DEBUG", "true").lower() in ("true", "1"),
            host=os.getenv("HOST", "0.0.0.0"),
            port=int(os.getenv("PORT", "8000")),
            workers=int(os.getenv("WORKERS", "1")),
            db_url=os.getenv("DATABASE_URL", "sqlite:///dev.db"),
            redis_url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
            secret_key=os.getenv("SECRET_KEY", "dev-secret-key"),
            log_level=os.getenv("LOG_LEVEL", "DEBUG"),
        )

# 不同环境的配置
# development: .env.dev
# production:  环境变量由运维/CI设置
# Docker:      -e KEY=VALUE 或 docker-compose environment

🔑 本课要点