#FastAPI Pydantic Settings多环境配置完全指南

# ❌ 错误示例：硬编码配置
class BadConfig:
    DATABASE_URL = "postgresql://localhost:5432/mydb_dev"  # 开发数据库
    SECRET_KEY = "dev-secret-key"                          # 开发密钥
    DEBUG = True                                         # 开发模式
    REDIS_URL = "redis://localhost:6379/0"              # 本地Redis
    
    # 上线时手动修改？容易遗漏或错误！
    # 安全风险：敏感信息暴露在代码中
    # 维护困难：多个环境配置混杂

# ✅ 正确示例：环境驱动的配置管理
"""
开发环境 (.env.development)    测试环境 (.env.test)      生产环境 (.env.production)
────────────────────────────  ────────────────────────  ────────────────────────────
DATABASE_URL=sqlite://...     DATABASE_URL=sqlite://...  DATABASE_URL=postgresql://...
SECRET_KEY=dev-key            SECRET_KEY=test-key       SECRET_KEY=production-key
DEBUG=true                   DEBUG=false               DEBUG=false
LOG_LEVEL=debug              LOG_LEVEL=info            LOG_LEVEL=warning
REDIS_URL=redis://...        REDIS_URL=redis://...     REDIS_URL=redis://cluster...
"""

# config/base.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, validator, field_validator
from typing import List, Optional
from pathlib import Path
import os

class BaseConfig(BaseSettings):
    """基础配置类 - 所有环境配置的基类"""
    
    model_config = SettingsConfigDict(
        env_file=".env",                    # 默认读取 .env 文件
        env_file_encoding="utf-8",         # 文件编码
        case_sensitive=False,              # 环境变量大小写不敏感
        extra="forbid",                    # 禁止额外字段
        protected_namespaces=("settings_",) # 保护命名空间
    )
    
    # 应用基本信息
    app_name: str = Field(
        default="FastAPI Application",
        description="应用名称",
        min_length=1,
        max_length=100
    )
    app_version: str = Field(
        default="1.0.0",
        description="应用版本号"
    )
    environment: str = Field(
        default="development",
        description="运行环境",
        pattern=r"^(development|testing|staging|production)$"
    )
    
    # 调试与开发
    debug: bool = Field(
        default=False,
        description="是否启用调试模式"
    )
    reload: bool = Field(
        default=False,
        description="是否启用热重载（开发环境）"
    )
    
    # 服务器配置
    host: str = Field(
        default="127.0.0.1",
        description="服务器主机地址"
    )
    port: int = Field(
        default=8000,
        ge=1,                                    # 大于等于1
        le=65535,                               # 小于等于65535
        description="服务器端口号"
    )
    
    # CORS配置
    cors_allow_origins: List[str] = Field(
        default_factory=list,
        description="允许的CORS来源"
    )
    cors_allow_credentials: bool = Field(
        default=True,
        description="是否允许CORS凭据"
    )
    cors_allow_methods: List[str] = Field(
        default_factory=lambda: ["*"],
        description="允许的CORS方法"
    )
    cors_allow_headers: List[str] = Field(
        default_factory=lambda: ["*"],
        description="允许的CORS头部"
    )
    
    # 日志配置
    log_level: str = Field(
        default="INFO",
        description="日志级别",
        pattern=r"^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$"
    )
    log_format: str = Field(
        default="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        description="日志格式"
    )
    
    # 会话配置
    session_secret_key: str = Field(
        default="change-me-in-production",
        description="会话密钥",
        min_length=32
    )
    
    @field_validator('cors_allow_origins', mode='before')
    @classmethod
    def parse_cors_origins(cls, v):
        """解析CORS来源列表"""
        if isinstance(v, str):
            return [origin.strip() for origin in v.split(',')]
        return v
    
    @field_validator('log_level')
    @classmethod
    def validate_log_level(cls, v):
        """验证日志级别"""
        import logging
        valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
        if v.upper() not in valid_levels:
            raise ValueError(f"Invalid log level: {v}. Must be one of {valid_levels}")
        return v.upper()
    
    @property
    def is_development(self) -> bool:
        """是否为开发环境"""
        return self.environment.lower() == "development"
    
    @property
    def is_testing(self) -> bool:
        """是否为测试环境"""
        return self.environment.lower() == "testing"
    
    @property
    def is_production(self) -> bool:
        """是否为生产环境"""
        return self.environment.lower() == "production"
    
    @property
    def is_staging(self) -> bool:
        """是否为预发布环境"""
        return self.environment.lower() == "staging"

# 使用示例
if __name__ == "__main__":
    config = BaseConfig()
    print(f"App: {config.app_name} v{config.app_version}")
    print(f"Environment: {config.environment}")
    print(f"Debug: {config.debug}")
    print(f"Host: {config.host}:{config.port}")

# config/environments.py
from .base import BaseConfig
from pydantic import Field, PostgresDsn, RedisDsn
from typing import Optional
import os

class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    
    environment: str = "development"
    debug: bool = True
    reload: bool = True
    log_level: str = "DEBUG"
    
    # 数据库配置
    database_url: str = Field(
        default="sqlite+aiosqlite:///./dev_database.db",
        description="开发环境数据库URL"
    )
    
    # Redis配置
    redis_url: str = Field(
        default="redis://localhost:6379/0",
        description="开发环境Redis URL"
    )
    
    # CORS配置 - 开发环境允许所有来源
    cors_allow_origins: list = Field(
        default_factory=lambda: ["http://localhost:3000", "http://127.0.0.1:3000", "*"]
    )
    
    # 会话配置
    session_secret_key: str = Field(
        default="dev-session-secret-key-change-in-production",
        min_length=32
    )
    
    # 邮件配置
    smtp_server: str = Field(default="localhost")
    smtp_port: int = Field(default=1025)  # 开发邮件服务器端口
    smtp_username: str = Field(default="")
    smtp_password: str = Field(default="")


class TestingConfig(BaseConfig):
    """测试环境配置"""
    
    environment: str = "testing"
    debug: bool = True
    testing: bool = True
    log_level: str = "WARNING"
    
    # 测试数据库
    database_url: str = Field(
        default="sqlite+aiosqlite:///./test_database.db",
        description="测试环境数据库URL"
    )
    
    # 测试Redis
    redis_url: str = Field(
        default="redis://localhost:6379/1",
        description="测试环境Redis URL"
    )
    
    # 测试时的CORS配置
    cors_allow_origins: list = Field(
        default_factory=lambda: ["http://localhost:3000"]
    )
    
    # 禁用邮件发送
    smtp_enabled: bool = Field(default=False)
    
    # 测试特定配置
    test_database_cleanup: bool = Field(default=True)
    test_data_seeding: bool = Field(default=True)


class StagingConfig(BaseConfig):
    """预发布环境配置"""
    
    environment: str = "staging"
    debug: bool = False
    log_level: str = "INFO"
    
    # 预发布数据库
    database_url: str = Field(
        default="postgresql+asyncpg://user:pass@staging-db:5432/staging_app",
        description="预发布环境数据库URL"
    )
    
    # 预发布Redis
    redis_url: str = Field(
        default="redis://staging-redis:6379/0",
        description="预发布环境Redis URL"
    )
    
    # 预发布CORS配置
    cors_allow_origins: list = Field(
        default_factory=lambda: ["https://staging.yourapp.com"]
    )
    
    # 安全配置
    session_secret_key: str = Field(min_length=32)
    csrf_enabled: bool = Field(default=True)


class ProductionConfig(BaseConfig):
    """生产环境配置"""
    
    environment: str = "production"
    debug: bool = False
    reload: bool = False
    log_level: str = "WARNING"
    
    # 生产数据库 - 必须通过环境变量提供
    database_url: str = Field(
        description="生产环境数据库URL",
        min_length=10
    )
    
    # 生产Redis
    redis_url: str = Field(
        description="生产环境Redis URL",
        min_length=10
    )
    
    # 生产CORS配置 - 严格的域名限制
    cors_allow_origins: list = Field(
        default_factory=lambda: ["https://yourapp.com", "https://www.yourapp.com"]
    )
    
    # 生产安全配置
    session_secret_key: str = Field(min_length=64)
    csrf_enabled: bool = Field(default=True)
    rate_limit_enabled: bool = Field(default=True)
    rate_limit_requests: int = Field(default=100)
    rate_limit_window: int = Field(default=60)  # 60秒窗口
    
    # 监控和告警
    sentry_dsn: Optional[str] = Field(default=None)
    monitoring_enabled: bool = Field(default=True)
    
    # SSL/TLS配置
    ssl_enabled: bool = Field(default=True)
    force_https_redirects: bool = Field(default=True)

    @field_validator('database_url')
    @classmethod
    def validate_production_database_url(cls, v):
        """验证生产环境数据库URL"""
        if not v or v == "sqlite+aiosqlite:///./prod_database.db":
            raise ValueError("Production environment must use a proper database URL, not SQLite")
        return v

# 配置工厂函数
def get_config() -> BaseConfig:
    """根据环境变量获取相应的配置"""
    env = os.getenv("ENVIRONMENT", os.getenv("FLASK_ENV", "development")).lower()
    
    config_map = {
        "development": DevelopmentConfig,
        "testing": TestingConfig,
        "staging": StagingConfig,
        "production": ProductionConfig,
    }
    
    config_class = config_map.get(env, DevelopmentConfig)
    return config_class()

# 使用示例
config = get_config()
print(f"Running in {config.environment} mode")
print(f"Database URL: {config.database_url}")

# config/manager.py
from .environments import get_config, BaseConfig
from functools import lru_cache
import logging
from typing import Dict, Any
import json

class ConfigManager:
    """配置管理器 - 提供配置的统一访问接口"""
    
    def __init__(self):
        self._config = None
        self._logger = logging.getLogger(__name__)
    
    @property
    def config(self) -> BaseConfig:
        """获取配置实例（单例模式）"""
        if self._config is None:
            self._config = get_config()
            self._logger.info(f"Loaded {self._config.environment} configuration")
        return self._config
    
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置值"""
        try:
            return getattr(self.config, key, default)
        except AttributeError:
            return default
    
    def set(self, key: str, value: Any) -> None:
        """设置配置值（仅在开发环境允许）"""
        if self.config.is_development:
            setattr(self.config, key, value)
        else:
            raise RuntimeError("Cannot modify configuration in non-development environments")
    
    def reload(self) -> None:
        """重新加载配置"""
        self._config = get_config()
        self._logger.info("Configuration reloaded")
    
    def dump(self) -> Dict[str, Any]:
        """导出配置（过滤敏感信息）"""
        config_dict = self.config.model_dump()
        
        # 过滤敏感字段
        sensitive_fields = [
            'session_secret_key', 'smtp_password', 'database_url', 
            'redis_url', 'sentry_dsn'
        ]
        
        for field in sensitive_fields:
            if field in config_dict:
                config_dict[field] = "***REDACTED***"
        
        return config_dict
    
    def validate(self) -> bool:
        """验证配置完整性"""
        try:
            # 尝试重新创建配置实例以验证
            temp_config = get_config()
            return True
        except Exception as e:
            self._logger.error(f"Configuration validation failed: {e}")
            return False

# 全局配置管理器实例
config_manager = ConfigManager()

# 使用装饰器缓存配置
@lru_cache(maxsize=1)
def get_cached_config() -> BaseConfig:
    """获取缓存的配置实例"""
    return get_config()

# 在FastAPI应用中使用
"""
from fastapi import FastAPI
from config.manager import config_manager

app = FastAPI(
    title=config_manager.config.app_name,
    version=config_manager.config.app_version,
    debug=config_manager.config.debug
)

@app.get("/health/config")
async def get_config_info():
    return {
        "environment": config_manager.config.environment,
        "debug": config_manager.config.debug,
        "database_url_set": bool(config_manager.config.database_url),
        "redis_url_set": bool(config_manager.config.redis_url)
    }
"""

# config/priority_example.py
import os
from pydantic_settings import BaseSettings, SettingsConfigDict

class PriorityConfig(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    app_name: str = "Default App Name"           # 最低优先级：默认值
    debug: bool = False                         # 最低优先级：默认值
    custom_setting: str = "default_value"      # 最低优先级：默认值

# 设置环境变量（第二高优先级）
os.environ['APP_NAME'] = 'Env Variable App'

# 创建配置实例时传入参数（最高优先级）
config = PriorityConfig(
    app_name="Parameter App",      # 最高优先级：直接传递参数
    debug=True                     # 最高优先级：直接传递参数
)

print(f"App Name: {config.app_name}")           # 输出: Parameter App
print(f"Debug: {config.debug}")                 # 输出: True
print(f"Custom Setting: {config.custom_setting}") # 输出: default_value (来自.env或默认值)

# config/file_loader.py
import os
from pathlib import Path
from typing import Optional
from pydantic_settings import BaseSettings, SettingsConfigDict

class EnvironmentSpecificConfig(BaseSettings):
    """环境特定配置加载器"""
    
    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type["EnvironmentSpecificConfig"],
        init_settings,
        env_settings,
        dotenv_settings,
        file_secret_settings,
    ):
        """自定义配置源加载顺序"""
        environment = os.getenv("ENVIRONMENT", "development")
        
        # 定义配置文件路径
        config_files = [
            f".env.{environment}",    # 环境特定的 .env 文件
            ".env",                  # 默认 .env 文件
            f"config/{environment}.json",  # 环境特定JSON配置
            "config/default.json",         # 默认JSON配置
        ]
        
        # 过滤存在的文件
        existing_files = [
            file_path for file_path in config_files 
            if Path(file_path).exists()
        ]
        
        # 自定义加载顺序：环境特定 > 默认 > 环境变量 > 初始化设置
        return (
            init_settings,           # 直接传递的参数（最高优先级）
            env_settings,            # 环境变量
            *(dotenv_settings for _ in existing_files if _.endswith('.env')),  # .env 文件
            *(lambda: load_json_file(f) for f in existing_files if f.endswith('.json')),  # JSON 文件
            file_secret_settings,    # 密钥文件
        )

def load_json_file(file_path: str) -> dict:
    """加载JSON配置文件"""
    import json
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

# 使用示例
"""
# 项目结构
project/
├── .env.development
├── .env.production
├── config/
│   ├── development.json
│   ├── production.json
│   └── default.json
└── app.py
"""

# config/validation.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, validator, field_validator, model_validator
from typing import List, Optional, Union
from urllib.parse import urlparse
import re

class ValidatedConfig(BaseSettings):
    """带高级验证的配置类"""
    
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    # 应用配置
    app_name: str = Field(
        default="FastAPI App",
        min_length=3,
        max_length=50,
        pattern=r'^[a-zA-Z0-9_-]+$',
        description="应用名称，只能包含字母、数字、下划线和连字符"
    )
    
    # 版本验证
    app_version: str = Field(
        default="1.0.0",
        pattern=r'^\d+\.\d+\.\d+(-[a-zA-Z0-9]+)?$',
        description="语义化版本号，格式: major.minor.patch[-suffix]"
    )
    
    # 环境验证
    environment: str = Field(
        default="development",
        pattern=r'^(development|testing|staging|production)$',
        description="运行环境"
    )
    
    # 数据库URL验证
    database_url: str = Field(
        description="数据库连接URL",
        min_length=10
    )
    
    # Redis URL验证
    redis_url: str = Field(
        default="redis://localhost:6379/0",
        description="Redis连接URL"
    )
    
    # 端口验证
    port: int = Field(
        default=8000,
        ge=1,
        le=65535,
        description="服务器端口"
    )
    
    # CORS来源验证
    cors_allow_origins: List[str] = Field(
        default_factory=list,
        description="允许的CORS来源"
    )
    
    # 超时验证
    request_timeout: int = Field(
        default=30,
        ge=1,
        le=300,
        description="请求超时时间（秒）"
    )
    
    # 缓存大小验证
    cache_max_items: int = Field(
        default=1000,
        ge=1,
        le=100000,
        description="缓存最大项目数"
    )
    
    # 密钥长度验证
    session_secret_key: str = Field(
        description="会话密钥",
        min_length=32
    )
    
    # 自定义字段验证器
    @field_validator('database_url')
    @classmethod
    def validate_database_url(cls, v):
        """验证数据库URL格式"""
        if not v:
            raise ValueError('Database URL cannot be empty')
        
        parsed = urlparse(v)
        valid_schemes = [
            'sqlite', 'postgresql', 'mysql', 'oracle', 
            'mssql', 'postgresql+asyncpg', 'mysql+pymysql'
        ]
        
        if parsed.scheme not in valid_schemes:
            raise ValueError(f'Invalid database scheme: {parsed.scheme}')
        
        return v
    
    @field_validator('redis_url')
    @classmethod
    def validate_redis_url(cls, v):
        """验证Redis URL格式"""
        if not v:
            raise ValueError('Redis URL cannot be empty')
        
        parsed = urlparse(v)
        if parsed.scheme != 'redis':
            raise ValueError(f'Invalid Redis scheme: {parsed.scheme}')
        
        return v
    
    @field_validator('cors_allow_origins', mode='before')
    @classmethod
    def validate_cors_origins(cls, v):
        """验证CORS来源列表"""
        if isinstance(v, str):
            origins = [origin.strip() for origin in v.split(',') if origin.strip()]
        elif isinstance(v, list):
            origins = v
        else:
            origins = []
        
        # 验证每个来源URL
        for origin in origins:
            if origin != "*" and not cls._is_valid_url(origin):
                raise ValueError(f'Invalid CORS origin: {origin}')
        
        return origins
    
    @field_validator('session_secret_key')
    @classmethod
    def validate_secret_key_strength(cls, v):
        """验证密钥强度"""
        if len(v) < 32:
            raise ValueError('Session secret key must be at least 32 characters long')
        
        # 检查密钥复杂度（生产环境）
        if len(v) < 64:
            import warnings
            warnings.warn(
                "Session secret key should be at least 64 characters long for production",
                UserWarning
            )
        
        return v
    
    @model_validator(mode='after')
    def validate_production_constraints(self):
        """验证生产环境约束"""
        if self.environment == 'production':
            # 生产环境不能使用SQLite
            if 'sqlite' in self.database_url.lower():
                raise ValueError('Production environment cannot use SQLite database')
            
            # 生产环境必须启用HTTPS
            https_origins = [origin for origin in self.cors_allow_origins if origin.startswith('https://')]
            if len(https_origins) != len(self.cors_allow_origins) and '*' not in self.cors_allow_origins:
                raise ValueError('Production environment should only allow HTTPS origins')
        
        return self
    
    @staticmethod
    def _is_valid_url(url: str) -> bool:
        """验证URL格式"""
        pattern = re.compile(
            r'^https?://'  # http:// or https://
            r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|'  # domain...
            r'localhost|'  # localhost...
            r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'  # ...or ip
            r'(?::\d+)?'  # optional port
            r'(?:/?|[/?]\S+)$', re.IGNORECASE)
        return pattern.match(url) is not None

# 使用示例和测试
def test_validation():
    """测试配置验证"""
    try:
        # 有效配置
        valid_config = ValidatedConfig(
            app_name="MyApp",
            app_version="1.0.0",
            environment="development",
            database_url="sqlite:///./test.db",
            session_secret_key="a_very_long_secret_key_that_meets_the_minimum_length_requirement"
        )
        print("✓ Valid configuration created successfully")
        
        # 验证配置
        print(f"App: {valid_config.app_name}")
        print(f"Environment: {valid_config.environment}")
        print(f"Database: {valid_config.database_url}")
        
    except Exception as e:
        print(f"✗ Configuration validation failed: {e}")

if __name__ == "__main__":
    test_validation()

# config/typesafe.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, TypeAdapter
from typing import Literal, Union, Optional
from enum import Enum
import ipaddress
from datetime import timedelta

# 定义枚举类型
class EnvironmentEnum(str, Enum):
    DEVELOPMENT = "development"
    TESTING = "testing"
    STAGING = "staging"
    PRODUCTION = "production"

class LogLevelEnum(str, Enum):
    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    CRITICAL = "CRITICAL"

class DatabaseDriverEnum(str, Enum):
    SQLITE = "sqlite"
    POSTGRESQL = "postgresql"
    MYSQL = "mysql"
    ORACLE = "oracle"

class RateLimitStrategyEnum(str, Enum):
    FIXED_WINDOW = "fixed_window"
    SLIDING_WINDOW = "sliding_window"
    TOKEN_BUCKET = "token_bucket"

class TypeSafeConfig(BaseSettings):
    """类型安全的配置类"""
    
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    # 使用枚举类型的环境配置
    environment: EnvironmentEnum = Field(
        default=EnvironmentEnum.DEVELOPMENT,
        description="运行环境"
    )
    
    # 使用枚举类型的日志级别
    log_level: LogLevelEnum = Field(
        default=LogLevelEnum.INFO,
        description="日志级别"
    )
    
    # 使用字面量类型的数据库驱动
    database_driver: DatabaseDriverEnum = Field(
        default=DatabaseDriverEnum.SQLITE,
        description="数据库驱动类型"
    )
    
    # 使用联合类型的数据库URL
    database_url: Union[
        str,  # 传统字符串URL
    ] = Field(
        description="数据库连接URL"
    )
    
    # 使用字面量类型的限流策略
    rate_limit_strategy: RateLimitStrategyEnum = Field(
        default=RateLimitStrategyEnum.FIXED_WINDOW,
        description="限流策略"
    )
    
    # 使用timedelta类型的超时配置
    request_timeout: timedelta = Field(
        default=timedelta(seconds=30),
        description="请求超时时间"
    )
    
    # 使用IP地址类型的允许列表
    allowed_ips: Optional[str] = Field(
        default=None,
        description="允许的IP地址列表（逗号分隔）"
    )
    
    # 使用字面量类型的缓存策略
    cache_strategy: Literal["memory", "redis", "memcached"] = Field(
        default="memory",
        description="缓存策略"
    )
    
    # 使用泛型类型的自定义配置
    custom_settings: Optional[dict] = Field(
        default=None,
        description="自定义配置字典"
    )
    
    @property
    def is_debug_mode(self) -> bool:
        """是否为调试模式"""
        return self.log_level in [LogLevelEnum.DEBUG, LogLevelEnum.INFO]
    
    @property
    def database_host(self) -> Optional[str]:
        """从数据库URL中提取主机"""
        import re
        match = re.search(r'@([^:/]+)', self.database_url)
        return match.group(1) if match else None
    
    @property
    def database_port(self) -> Optional[int]:
        """从数据库URL中提取端口"""
        import re
        match = re.search(r':(\d+)', self.database_url)
        return int(match.group(1)) if match else None

# 配置适配器 - 用于运行时类型转换
DatabaseUrlAdapter = TypeAdapter(Union[str])

def get_typed_config() -> TypeSafeConfig:
    """获取类型安全的配置实例"""
    return TypeSafeConfig()

# 使用示例
def demonstrate_type_safety():
    """演示类型安全"""
    config = TypeSafeConfig(
        environment=EnvironmentEnum.PRODUCTION,
        log_level=LogLevelEnum.WARNING,
        database_driver=DatabaseDriverEnum.POSTGRESQL,
        database_url="postgresql://user:pass@localhost:5432/mydb",
        rate_limit_strategy=RateLimitStrategyEnum.SLIDING_WINDOW,
        request_timeout=timedelta(minutes=1),
        cache_strategy="redis"
    )
    
    print(f"Environment: {config.environment.value}")
    print(f"Log Level: {config.log_level.value}")
    print(f"Database Driver: {config.database_driver.value}")
    print(f"Rate Limit Strategy: {config.rate_limit_strategy.value}")
    print(f"Request Timeout: {config.request_timeout}")
    print(f"Is Debug Mode: {config.is_debug_mode}")
    print(f"Database Host: {config.database_host}")
    print(f"Database Port: {config.database_port}")

if __name__ == "__main__":
    demonstrate_type_safety()

# config/secrets.py
import os
import secrets
import hashlib
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from typing import Optional
import json
from pathlib import Path

class SecretManager:
    """敏感信息安全管理器"""
    
    def __init__(self, master_password: Optional[str] = None):
        self.master_password = master_password or os.getenv('MASTER_PASSWORD')
        self._fernet = None
        self.secrets_dir = Path("secrets")
        self.secrets_dir.mkdir(exist_ok=True)
    
    @property
    def fernet(self) -> Fernet:
        """获取加密器实例"""
        if self._fernet is None:
            if not self.master_password:
                raise ValueError("Master password not provided")
            self._fernet = self._create_fernet_instance()
        return self._fernet
    
    def _create_fernet_instance(self) -> Fernet:
        """创建Fernet加密实例"""
        password = self.master_password.encode()
        salt = b'secret_salt_for_fernet'  # 在实际应用中应该从安全的地方获取
        
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password))
        return Fernet(key)
    
    def encrypt_secret(self, secret: str) -> str:
        """加密敏感信息"""
        encrypted_bytes = self.fernet.encrypt(secret.encode())
        return base64.b64encode(encrypted_bytes).decode()
    
    def decrypt_secret(self, encrypted_secret: str) -> str:
        """解密敏感信息"""
        encrypted_bytes = base64.b64decode(encrypted_secret.encode())
        decrypted_bytes = self.fernet.decrypt(encrypted_bytes)
        return decrypted_bytes.decode()
    
    def store_secret(self, key: str, secret: str, encrypt: bool = True) -> Path:
        """存储敏感信息"""
        file_path = self.secrets_dir / f"{key}.secret"
        
        if encrypt:
            secret = self.encrypt_secret(secret)
        
        with open(file_path, 'w') as f:
            f.write(secret)
        
        # 设置安全的文件权限
        os.chmod(file_path, 0o600)  # 只有所有者可读写
        
        return file_path
    
    def load_secret(self, key: str, decrypt: bool = True) -> Optional[str]:
        """加载敏感信息"""
        file_path = self.secrets_dir / f"{key}.secret"
        
        if not file_path.exists():
            return None
        
        with open(file_path, 'r') as f:
            secret = f.read().strip()
        
        if decrypt:
            secret = self.decrypt_secret(secret)
        
        return secret
    
    def generate_secure_token(self, length: int = 32) -> str:
        """生成安全的随机令牌"""
        return secrets.token_urlsafe(length)
    
    def generate_strong_password(self, length: int = 64) -> str:
        """生成强密码"""
        alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%^&*"
        return ''.join(secrets.choice(alphabet) for _ in range(length))

# 全局密钥管理器实例
secret_manager = SecretManager()

# 使用示例
"""
# 存储数据库密码
db_password = "my_secure_db_password"
secret_manager.store_secret("database_password", db_password)

# 加载数据库密码
stored_password = secret_manager.load_secret("database_password")
print(f"Stored password: {stored_password}")
"""

# config/env_security.py
import os
import re
from typing import Dict, List, Optional
import logging

class EnvironmentSecurityManager:
    """环境变量安全管理器"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.sensitive_keys = {
            'PASSWORD', 'SECRET', 'TOKEN', 'KEY', 'AUTH', 'CREDENTIAL',
            'PRIVATE', 'CLIENT_ID', 'CLIENT_SECRET', 'API_KEY', 'JWT_SECRET'
        }
        self.required_keys = set()
        self.optional_keys = set()
    
    def register_required_vars(self, *keys: str) -> None:
        """注册必需的环境变量"""
        self.required_keys.update(keys)
    
    def register_optional_vars(self, *keys: str) -> None:
        """注册可选的环境变量"""
        self.optional_keys.update(keys)
    
    def validate_environment_vars(self) -> List[str]:
        """验证环境变量"""
        errors = []
        
        # 检查必需变量
        for key in self.required_keys:
            if not os.getenv(key):
                errors.append(f"Required environment variable '{key}' is not set")
        
        # 检查敏感变量的强度
        for key in os.environ:
            if self._is_sensitive_key(key):
                value = os.getenv(key, "")
                if value and not self._is_value_secure(value):
                    errors.append(f"Environment variable '{key}' does not meet security requirements")
        
        return errors
    
    def mask_sensitive_values(self, data: Dict) -> Dict:
        """屏蔽敏感信息"""
        masked_data = {}
        for key, value in data.items():
            if self._is_sensitive_key(key):
                masked_data[key] = "***REDACTED***"
            else:
                masked_data[key] = value
        return masked_data
    
    def get_secure_env_var(self, key: str, default: str = "", min_length: int = 0) -> str:
        """安全获取环境变量"""
        value = os.getenv(key, default)
        
        if min_length > 0 and len(value) < min_length:
            raise ValueError(f"Environment variable '{key}' must be at least {min_length} characters long")
        
        if self._is_sensitive_key(key) and not self._is_value_secure(value):
            self.logger.warning(f"Sensitive environment variable '{key}' may not meet security requirements")
        
        return value
    
    def _is_sensitive_key(self, key: str) -> bool:
        """检查是否为敏感键"""
        key_upper = key.upper()
        return any(sensitive in key_upper for sensitive in self.sensitive_keys)
    
    def _is_value_secure(self, value: str) -> bool:
        """检查值是否安全"""
        if not value:
            return True  # 空值被认为是安全的
        
        # 检查最小长度
        if len(value) < 8:
            return False
        
        # 检查是否为常见弱密码
        weak_patterns = [
            r'^(password|admin|123456|qwerty)$',
            r'^(secret|key|token)$',
            r'^(change|default|test)$'
        ]
        
        for pattern in weak_patterns:
            if re.match(pattern, value.lower()):
                return False
        
        # 检查复杂度（如果有要求）
        has_lower = any(c.islower() for c in value)
        has_upper = any(c.isupper() for c in value)
        has_digit = any(c.isdigit() for c in value)
        has_special = any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in value)
        
        complexity_score = sum([has_lower, has_upper, has_digit, has_special])
        return complexity_score >= 3  # 至少3种不同类型字符

# 全局环境安全管理器
env_security_manager = EnvironmentSecurityManager()

# 在配置类中使用
"""
class SecureConfig(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    database_url: str
    jwt_secret: str = Field(min_length=32)
    
    def __init__(self, **kwargs):
        # 验证环境变量
        errors = env_security_manager.validate_environment_vars()
        if errors:
            raise ValueError(f"Environment validation failed: {errors}")
        
        super().__init__(**kwargs)
    
    @field_validator('jwt_secret')
    @classmethod
    def validate_jwt_secret(cls, v):
        if not env_security_manager._is_value_secure(v):
            raise ValueError("JWT secret does not meet security requirements")
        return v
"""

# config/docker_secrets.py
import os
from pathlib import Path
from typing import Optional
import json

class DockerSecretsManager:
    """Docker Secrets管理器"""
    
    def __init__(self, secrets_dir: str = "/run/secrets"):
        self.secrets_dir = Path(secrets_dir)
        self._cache = {}
    
    def get_secret(self, name: str, default: Optional[str] = None) -> Optional[str]:
        """从Docker Secrets获取值"""
        if name in self._cache:
            return self._cache[name]
        
        secret_path = self.secrets_dir / name
        
        if secret_path.exists():
            try:
                with open(secret_path, 'r') as f:
                    value = f.read().strip()
                    self._cache[name] = value
                    return value
            except IOError:
                pass
        
        # 如果在Kubernetes中，可能需要从环境变量获取
        env_var_name = f"{name.upper()}_FILE"
        file_path = os.getenv(env_var_name)
        if file_path:
            try:
                with open(file_path, 'r') as f:
                    value = f.read().strip()
                    self._cache[name] = value
                    return value
            except IOError:
                pass
        
        return default
    
    def get_secret_as_json(self, name: str, default: Optional[dict] = None) -> Optional[dict]:
        """将Docker Secret作为JSON获取"""
        secret_content = self.get_secret(name)
        if secret_content:
            try:
                return json.loads(secret_content)
            except json.JSONDecodeError:
                raise ValueError(f"Docker secret '{name}' is not valid JSON")
        return default
    
    def get_secret_as_int(self, name: str, default: Optional[int] = None) -> Optional[int]:
        """将Docker Secret作为整数获取"""
        secret_content = self.get_secret(name)
        if secret_content:
            try:
                return int(secret_content)
            except ValueError:
                raise ValueError(f"Docker secret '{name}' is not a valid integer")
        return default
    
    def get_secret_as_bool(self, name: str, default: Optional[bool] = None) -> Optional[bool]:
        """将Docker Secret作为布尔值获取"""
        secret_content = self.get_secret(name)
        if secret_content:
            return secret_content.lower() in ('true', '1', 'yes', 'on')
        return default

# 全局Docker Secrets管理器
docker_secrets_manager = DockerSecretsManager()

# 在配置中使用Docker Secrets
"""
class DockerConfig(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    # 从Docker Secrets获取数据库密码
    database_password: Optional[str] = None
    jwt_secret: Optional[str] = None
    
    def __init__(self, **kwargs):
        # 如果没有通过环境变量设置，则尝试从Docker Secrets获取
        if 'database_password' not in kwargs:
            kwargs['database_password'] = docker_secrets_manager.get_secret('db_password')
        if 'jwt_secret' not in kwargs:
            kwargs['jwt_secret'] = docker_secrets_manager.get_secret('jwt_secret')
        
        super().__init__(**kwargs)
"""