import os

# 基礎(chǔ)用法：讀取環(huán)境變量
DATABASE_URL = os.environ["DATABASE_URL"]  # 不存在則拋出 KeyError
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///dev.db")  # 提供默認值

# 類型轉(zhuǎn)換：環(huán)境變量都是字符串
DEBUG = os.environ.get("DEBUG", "false").lower() == "true"
PORT = int(os.environ.get("PORT", "8000"))
MAX_WORKERS = int(os.environ.get("MAX_WORKERS", "4"))

pip install python-dotenv

# .env（絕對不能提交到 Git?。?
DATABASE_URL=postgresql://user:password@localhost:5432/devdb
REDIS_URL=redis://localhost:6379/0
SECRET_KEY=your-local-secret-key-never-use-in-production
DEBUG=true
LOG_LEVEL=DEBUG

.env
.env.local
.env.*.local

from dotenv import load_dotenv
import os

# 加載 .env 文件（生產(chǎn)環(huán)境中 .env 文件不存在，不影響運行）
load_dotenv()

DATABASE_URL = os.environ["DATABASE_URL"]
DEBUG = os.environ.get("DEBUG", "false").lower() == "true"

# .env.example — 復(fù)制為 .env 并填寫真實值
DATABASE_URL=postgresql://user:password@host:5432/dbname
REDIS_URL=redis://host:6379/0
SECRET_KEY=your-secret-key-here
DEBUG=false
LOG_LEVEL=INFO

# 嘗試用環(huán)境變量表達嵌套配置——丑陋且難以維護
DATABASE_PRIMARY_HOST=db1.example.com
DATABASE_PRIMARY_PORT=5432
DATABASE_REPLICA_HOST=db2.example.com
DATABASE_REPLICA_PORT=5432

# 所有環(huán)境共享的默認配置
app:
  name: "My Application"
  version: "1.0.0"
  timezone: "Asia/Shanghai"

server:
  host: "0.0.0.0"
  port: 8000
  workers: 4

logging:
  level: "INFO"
  format: "json"
  
cache:
  ttl: 3600
  max_size: 1000

# 僅覆蓋開發(fā)環(huán)境特有的配置
server:
  port: 8080
  
logging:
  level: "DEBUG"
  format: "console"  # 開發(fā)環(huán)境用人類可讀格式

database:
  url: "postgresql://dev:dev@localhost:5432/myapp_dev"
  pool_size: 2
  echo: true  # 打印 SQL 語句

cache:
  ttl: 60  # 開發(fā)環(huán)境緩存時間短

server:
  workers: 16  # 生產(chǎn)環(huán)境更多 worker

logging:
  level: "WARNING"  # 生產(chǎn)環(huán)境減少日志量

database:
  pool_size: 20
  pool_timeout: 30
  echo: false

import yaml
import os
from pathlib import Path
from typing import Any


def deep_merge(base: dict, override: dict) -> dict:
    """
    深度合并兩個字典，override 的值會覆蓋 base 的值
    支持嵌套字典的遞歸合并
    """
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result


def load_config() -> dict:
    """
    按優(yōu)先級加載配置：
    base.yaml < {env}.yaml < 環(huán)境變量
    """
    config_dir = Path(__file__).parent / "config"
    env = os.environ.get("APP_ENV", "development")
    
    # 1. 加載基礎(chǔ)配置
    base_path = config_dir / "base.yaml"
    with open(base_path) as f:
        config = yaml.safe_load(f)
    
    # 2. 加載環(huán)境特定配置（覆蓋基礎(chǔ)配置）
    env_path = config_dir / f"{env}.yaml"
    if env_path.exists():
        with open(env_path) as f:
            env_config = yaml.safe_load(f) or {}
        config = deep_merge(config, env_config)
    
    # 3. 環(huán)境變量中的配置優(yōu)先級最高
    # 支持 APP__DATABASE__URL 形式的嵌套鍵
    for key, value in os.environ.items():
        if key.startswith("APP__"):
            keys = key[5:].lower().split("__")
            nested = config
            for k in keys[:-1]:
                nested = nested.setdefault(k, {})
            nested[keys[-1]] = value
    
    return config


# 全局配置單例
_config = None

def get_config() -> dict:
    global _config
    if _config is None:
        _config = load_config()
    return _config

from config_loader import get_config

config = get_config()

# 訪問配置
db_url = config["database"]["url"]
log_level = config["logging"]["level"]
server_port = config["server"]["port"]

from pydantic import BaseModel, validator, Field
from pydantic_settings import BaseSettings
from typing import Optional
import os


class DatabaseConfig(BaseModel):
    url: str
    pool_size: int = 10
    pool_timeout: int = 30
    echo: bool = False
    
    @validator("pool_size")
    def validate_pool_size(cls, v):
        if v < 1 or v > 100:
            raise ValueError("pool_size 必須在 1-100 之間")
        return v


class ServerConfig(BaseModel):
    host: str = "0.0.0.0"
    port: int = Field(8000, ge=1, le=65535)
    workers: int = Field(4, ge=1)


class LoggingConfig(BaseModel):
    level: str = "INFO"
    format: str = "json"
    
    @validator("level")
    def validate_level(cls, v):
        valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
        if v.upper() not in valid_levels:
            raise ValueError(f"日志級別必須是 {valid_levels} 之一")
        return v.upper()


class AppConfig(BaseSettings):
    """
    頂層配置模型
    自動從環(huán)境變量讀?。▋?yōu)先級高于默認值）
    """
    env: str = Field("development", env="APP_ENV")
    
    database: DatabaseConfig
    server: ServerConfig = ServerConfig()
    logging: LoggingConfig = LoggingConfig()
    
    class Config:
        env_prefix = "APP_"
        env_nested_delimiter = "__"


# 使用方式
def create_config() -> AppConfig:
    raw_config = load_config()  # 從 YAML 文件加載
    return AppConfig(**raw_config)


# 全局配置
settings = create_config()

# 類型安全的訪問
print(settings.database.url)       # str
print(settings.server.port)        # int，不需要類型轉(zhuǎn)換
print(settings.logging.level)      # 已驗證為合法值

import hvac
import os
from functools import lru_cache


class VaultConfigProvider:
    """從 HashiCorp Vault 讀取敏感配置"""
    
    def __init__(self):
        self.client = hvac.Client(
            url=os.environ["VAULT_ADDR"],
            token=os.environ["VAULT_TOKEN"]
        )
    
    def get_secret(self, path: str) -> dict:
        """讀取 KV v2 格式的密鑰"""
        response = self.client.secrets.kv.v2.read_secret_version(
            path=path,
            mount_point="secret"
        )
        return response["data"]["data"]
    
    def get_database_credentials(self) -> dict:
        """獲取動態(tài)數(shù)據(jù)庫憑證（每次調(diào)用都獲取新的臨時憑證）"""
        response = self.client.secrets.database.generate_credentials(
            name="my-app-role"
        )
        return {
            "username": response["data"]["username"],
            "password": response["data"]["password"],
            "lease_id": response["lease_id"],
            "lease_duration": response["lease_duration"]
        }


@lru_cache(maxsize=None)
def get_vault_provider() -> VaultConfigProvider:
    return VaultConfigProvider()


# 在應(yīng)用啟動時獲取敏感配置
vault = get_vault_provider()

# 獲取數(shù)據(jù)庫密碼（不再硬編碼）
db_secret = vault.get_secret("myapp/database")
DATABASE_URL = f"postgresql://{db_secret['username']}:{db_secret['password']}@{db_secret['host']}/myapp"

import redis
import json
import threading
from typing import Any, Callable


class DynamicConfig:
    """
    基于 Redis 的動態(tài)配置，支持熱更新
    
    使用 Redis Pub/Sub 監(jiān)聽配置變更
    """
    
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
        self._cache = {}
        self._listeners: dict[str, list[Callable]] = {}
        self._start_listener()
    
    def get(self, key: str, default: Any = None) -> Any:
        """獲取配置值（優(yōu)先從本地緩存讀?。?""
        if key not in self._cache:
            value = self.redis.get(f"config:{key}")
            if value:
                self._cache[key] = json.loads(value)
            else:
                return default
        return self._cache.get(key, default)
    
    def set(self, key: str, value: Any) -> None:
        """設(shè)置配置值，并通知所有實例"""
        self.redis.set(f"config:{key}", json.dumps(value))
        self.redis.publish("config:changes", json.dumps({"key": key, "value": value}))
    
    def on_change(self, key: str, callback: Callable) -> None:
        """注冊配置變更回調(diào)"""
        if key not in self._listeners:
            self._listeners[key] = []
        self._listeners[key].append(callback)
    
    def _start_listener(self) -> None:
        """在后臺線程監(jiān)聽配置變更"""
        def listen():
            pubsub = self.redis.pubsub()
            pubsub.subscribe("config:changes")
            for message in pubsub.listen():
                if message["type"] == "message":
                    data = json.loads(message["data"])
                    key = data["key"]
                    self._cache[key] = data["value"]
                    # 觸發(fā)回調(diào)
                    for callback in self._listeners.get(key, []):
                        callback(data["value"])
        
        thread = threading.Thread(target=listen, daemon=True)
        thread.start()


# 使用示例
dynamic_config = DynamicConfig(redis_url="redis://localhost:6379/1")

# 注冊限流閾值變更回調(diào)
def on_rate_limit_change(new_value):
    print(f"限流閾值已更新為: {new_value} 次/分鐘")
    # 實時更新限流器配置

dynamic_config.on_change("rate_limit_per_minute", on_rate_limit_change)

# 讀取動態(tài)配置
rate_limit = dynamic_config.get("rate_limit_per_minute", default=100)
feature_enabled = dynamic_config.get("feature_new_ui", default=False)

# 運維人員可以在不重啟服務(wù)的情況下調(diào)整
# dynamic_config.set("rate_limit_per_minute", 200)

import os
import yaml
from pathlib import Path
from functools import lru_cache
from pydantic import BaseModel, validator
from pydantic_settings import BaseSettings
from typing import Optional


class DatabaseSettings(BaseModel):
    url: str = "sqlite:///./dev.db"
    pool_size: int = 5
    echo: bool = False


class RedisSettings(BaseModel):
    url: str = "redis://localhost:6379/0"
    max_connections: int = 10


class AppSettings(BaseSettings):
    # 基礎(chǔ)信息
    app_name: str = "MyApp"
    app_env: str = "development"
    debug: bool = False
    secret_key: str = "change-this-in-production"
    
    # 子配置
    database: DatabaseSettings = DatabaseSettings()
    redis: RedisSettings = RedisSettings()
    
    # 功能開關(guān)
    feature_new_dashboard: bool = False
    
    class Config:
        env_file = ".env"
        env_prefix = "APP_"
        env_nested_delimiter = "__"
    
    @validator("secret_key")
    def validate_secret_key(cls, v, values):
        env = values.get("app_env", "development")
        if env == "production" and v == "change-this-in-production":
            raise ValueError("生產(chǎn)環(huán)境必須設(shè)置真實的 SECRET_KEY！")
        return v
    
    @validator("app_env")
    def validate_env(cls, v):
        allowed = {"development", "staging", "production", "test"}
        if v not in allowed:
            raise ValueError(f"app_env 必須是 {allowed} 之一")
        return v
    
    def is_production(self) -> bool:
        return self.app_env == "production"
    
    def is_development(self) -> bool:
        return self.app_env == "development"


@lru_cache(maxsize=1)
def get_settings() -> AppSettings:
    """
    獲取全局配置單例（使用 lru_cache 確保只初始化一次）
    在測試中可以通過 get_settings.cache_clear() 重置
    """
    return AppSettings()


# 在應(yīng)用的任何地方使用
settings = get_settings()

print(f"運行環(huán)境: {settings.app_env}")
print(f"數(shù)據(jù)庫: {settings.database.url}")
print(f"調(diào)試模式: {settings.debug}")

# tests/conftest.py
import pytest
from unittest.mock import patch
from config import get_settings, AppSettings


@pytest.fixture
def test_settings():
    """提供測試專用的配置"""
    test_config = AppSettings(
        app_env="test",
        debug=True,
        database={"url": "sqlite:///./test.db", "echo": True},
        secret_key="test-secret-key"
    )
    return test_config


@pytest.fixture(autouse=True)
def override_settings(test_settings):
    """自動替換所有測試中的配置"""
    # 清除 lru_cache 并替換
    get_settings.cache_clear()
    with patch("config.get_settings", return_value=test_settings):
        yield
    get_settings.cache_clear()