从环境变量到配置中心带你掌握Python多环境配置

import os

# 基础用法：读取环境变量
DATABASE_URL = os.environ["DATABASE_URL"]  # 不存在则抛出 KeyError
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///dev.db")  # 提供默认值

# 类型转换：环境变量都是字符串
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 文件（生产环境中 .env 文件不存在，不影响运行）
load_dotenv()

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

# .env.example — 复制为 .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

# 尝试用环境变量表达嵌套配置——丑陋且难以维护
DATABASE_PRIMARY_HOST=db1.example.com
DATABASE_PRIMARY_PORT=5432
DATABASE_REPLICA_HOST=db2.example.com
DATABASE_REPLICA_PORT=5432

# 所有环境共享的默认配置
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

# 仅覆盖开发环境特有的配置
server:
  port: 8080
  
logging:
  level: "DEBUG"
  format: "console"  # 开发环境用人类可读格式

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

cache:
  ttl: 60  # 开发环境缓存时间短

server:
  workers: 16  # 生产环境更多 worker

logging:
  level: "WARNING"  # 生产环境减少日志量

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:
    """
    按优先级加载配置：
    base.yaml < {env}.yaml < 环境变量
    """
    config_dir = Path(__file__).parent / "config"
    env = os.environ.get("APP_ENV", "development")
    
    # 1. 加载基础配置
    base_path = config_dir / "base.yaml"
    with open(base_path) as f:
        config = yaml.safe_load(f)
    
    # 2. 加载环境特定配置（覆盖基础配置）
    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. 环境变量中的配置优先级最高
    # 支持 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):
    """
    顶层配置模型
    自动从环境变量读取（优先级高于默认值）
    """
    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，不需要类型转换
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:
        """获取动态数据库凭证（每次调用都获取新的临时凭证）"""
        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()


# 在应用启动时获取敏感配置
vault = get_vault_provider()

# 获取数据库密码（不再硬编码）
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 的动态配置，支持热更新
    
    使用 Redis Pub/Sub 监听配置变更
    """
    
    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:
        """获取配置值（优先从本地缓存读取）"""
        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:
        """设置配置值，并通知所有实例"""
        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:
        """注册配置变更回调"""
        if key not in self._listeners:
            self._listeners[key] = []
        self._listeners[key].append(callback)
    
    def _start_listener(self) -> None:
        """在后台线程监听配置变更"""
        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"]
                    # 触发回调
                    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")

# 注册限流阈值变更回调
def on_rate_limit_change(new_value):
    print(f"限流阈值已更新为: {new_value} 次/分钟")
    # 实时更新限流器配置

dynamic_config.on_change("rate_limit_per_minute", on_rate_limit_change)

# 读取动态配置
rate_limit = dynamic_config.get("rate_limit_per_minute", default=100)
feature_enabled = dynamic_config.get("feature_new_ui", default=False)

# 运维人员可以在不重启服务的情况下调整
# 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):
    # 基础信息
    app_name: str = "MyApp"
    app_env: str = "development"
    debug: bool = False
    secret_key: str = "change-this-in-production"
    
    # 子配置
    database: DatabaseSettings = DatabaseSettings()
    redis: RedisSettings = RedisSettings()
    
    # 功能开关
    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("生产环境必须设置真实的 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()


# 在应用的任何地方使用
settings = get_settings()

print(f"运行环境: {settings.app_env}")
print(f"数据库: {settings.database.url}")
print(f"调试模式: {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()