Python 配置管理
大约 12 分钟约 3563 字
Python 配置管理
简介
配置管理是将代码与运行参数分离的工程实践，涵盖环境变量读取、配置文件解析、多环境管理和动态配置更新等方面。良好的配置管理确保应用在不同环境（开发/测试/生产）中安全、灵活地运行，而无需修改代码。
特点
实现
pydantic-settings 配置管理
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, SecretStr, field_validator
from typing import Literal
class AppSettings(BaseSettings):
"""应用配置（推荐方式）"""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
env_prefix="APP_", # 环境变量前缀
case_sensitive=False,
)
# 基础配置
app_name: str = "my-service"
debug: bool = False
environment: Literal["dev", "staging", "prod"] = "dev"
log_level: str = "INFO"
# 数据库配置
db_host: str = "localhost"
db_port: int = 5432
db_name: str = "mydb"
db_user: str = "postgres"
db_password: SecretStr = SecretStr("")
# Redis 配置
redis_url: str = "redis://localhost:6379/0"
# 服务配置
api_timeout: float = Field(default=30.0, ge=1.0, le=300.0)
max_workers: int = Field(default=4, ge=1, le=32)
cors_origins: list[str] = Field(default=["*"])
@property
def database_url(self) -> str:
return (
f"postgresql://{self.db_user}:{self.db_password.get_secret_value()}"
f"@{self.db_host}:{self.db_port}/{self.db_name}"
)
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
valid = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
upper = v.upper()
if upper not in valid:
raise ValueError(f"log_level 必须是 {valid} 之一，收到: {v}")
return upper
@property
def is_production(self) -> bool:
return self.environment == "prod"
# 使用
settings = AppSettings()
print(settings.database_url)
print(settings.is_production)
# 通过环境变量覆盖: APP_DB_HOST=prod-db.example.com APP_DEBUG=false多环境配置与 YAML 文件
import yaml
import os
from pathlib import Path
from typing import Any
from dataclasses import dataclass, field
@dataclass
class DatabaseConfig:
host: str = "localhost"
port: int = 5432
name: str = "mydb"
max_connections: int = 10
@dataclass
class CacheConfig:
backend: str = "redis"
url: str = "redis://localhost:6379/0"
ttl: int = 3600
@dataclass
class AppConfig:
env: str = "dev"
debug: bool = True
database: DatabaseConfig = field(default_factory=DatabaseConfig)
cache: CacheConfig = field(default_factory=CacheConfig)
features: dict[str, bool] = field(default_factory=dict)
class ConfigManager:
"""多环境配置管理器"""
def __init__(self, config_dir: str = "config"):
self.config_dir = Path(config_dir)
self._configs: dict[str, AppConfig] = {}
def load(self, env: str | None = None) -> AppConfig:
env = env or os.getenv("APP_ENV", "dev")
if env in self._configs:
return self._configs[env]
# 1. 加载基础配置
base_config = self._load_yaml(self.config_dir / "base.yaml")
# 2. 加载环境特定配置（覆盖基础配置）
env_file = self.config_dir / f"{env}.yaml"
env_config = self._load_yaml(env_file) if env_file.exists() else {}
# 3. 合并配置
merged = self._deep_merge(base_config, env_config)
# 4. 环境变量覆盖（最高优先级）
self._apply_env_overrides(merged)
config = self._dict_to_config(merged)
config.env = env
self._configs[env] = config
return config
def _load_yaml(self, path: Path) -> dict:
if not path.exists():
return {}
with open(path, "r", encoding="utf-8") as f:
return yaml.safe_load(f) or {}
def _deep_merge(self, base: dict, override: dict) -> dict:
result = base.copy()
for key, value in override.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._deep_merge(result[key], value)
else:
result[key] = value
return result
def _apply_env_overrides(self, config: dict) -> None:
prefix = "APP_"
for key, value in os.environ.items():
if key.startswith(prefix):
config_key = key[len(prefix):].lower()
config[config_key] = value
def _dict_to_config(self, d: dict) -> AppConfig:
db = d.pop("database", {})
cache = d.pop("cache", {})
return AppConfig(
**{k: v for k, v in d.items() if k in AppConfig.__dataclass_fields__},
database=DatabaseConfig(**db),
cache=CacheConfig(**cache),
)
# config/base.yaml:
# debug: false
# database:
# host: localhost
# port: 5432
# max_connections: 10
#
# config/prod.yaml:
# debug: false
# database:
# host: prod-db.internal
# max_connections: 50
# manager = ConfigManager()
# config = manager.load("prod").env 文件与动态配置刷新
from dotenv import load_dotenv, dotenv_values
import os
from typing import Any
import json
class EnvConfig:
"""基于 .env 文件的配置管理"""
def __init__(self, env_file: str = ".env", override: bool = False):
load_dotenv(env_file, override=override)
self._cache: dict[str, Any] = {}
def get(self, key: str, default: Any = None, cast: type | None = None) -> Any:
"""获取配置值，支持类型转换"""
if key in self._cache:
return self._cache[key]
value = os.getenv(key, default)
if value is None:
return default
if cast is not None:
value = self._cast(value, cast)
self._cache[key] = value
return value
def _cast(self, value: str, target_type: type) -> Any:
if target_type is bool:
return value.lower() in ("true", "1", "yes", "on")
if target_type is list:
return [item.strip() for item in value.split(",")]
return target_type(value)
@property
def is_debug(self) -> bool:
return self.get("DEBUG", False, bool)
@property
def database_url(self) -> str:
return self.get("DATABASE_URL", "sqlite:///default.db")
def require(self, key: str) -> Any:
"""获取必须存在的配置值"""
value = os.getenv(key)
if value is None:
raise ValueError(f"缺少必须的环境变量: {key}")
return value
def reload(self, env_file: str = ".env"):
"""重新加载配置"""
self._cache.clear()
load_dotenv(env_file, override=True)
# 使用
config = EnvConfig()
db_url = config.get("DATABASE_URL", "sqlite:///dev.db")
debug = config.get("DEBUG", False, bool)
allowed_hosts = config.get("ALLOWED_HOSTS", ["localhost"], list)
api_key = config.require("API_KEY") # 缺少则抛出异常配置验证与启动检查
from dataclasses import dataclass
from typing import Any
@dataclass
class ValidationResult:
is_valid: bool
errors: list[str]
class ConfigValidator:
"""配置启动前校验"""
def __init__(self, config: dict):
self.config = config
self.errors: list[str] = []
def require_keys(self, *keys: str) -> "ConfigValidator":
for key in keys:
if not self.config.get(key):
self.errors.append(f"缺少必须配置项: {key}")
return self
def require_url(self, key: str) -> "ConfigValidator":
value = self.config.get(key, "")
if value and not value.startswith(("http://", "https://", "redis://", "postgresql://")):
self.errors.append(f"{key} 不是有效的 URL: {value}")
return self
def require_range(self, key: str, min_val: int | float, max_val: int | float) -> "ConfigValidator":
value = self.config.get(key)
if value is not None:
if not (min_val <= value <= max_val):
self.errors.append(f"{key} 的值 {value} 不在范围 [{min_val}, {max_val}] 内")
return self
def validate(self) -> ValidationResult:
return ValidationResult(
is_valid=len(self.errors) == 0,
errors=self.errors.copy(),
)
# 使用
config = {
"database_url": "postgresql://prod-db:5432/mydb",
"redis_url": "redis://cache:6379",
"api_key": "",
"max_workers": 100,
}
result = (
ConfigValidator(config)
.require_keys("database_url", "redis_url", "api_key")
.require_url("database_url")
.require_url("redis_url")
.require_range("max_workers", 1, 32)
.validate()
)
if not result.is_valid:
for error in result.errors:
print(f"[配置错误] {error}")
raise SystemExit("配置校验失败，服务无法启动")配置热更新与文件监听
import os
import time
import logging
import hashlib
from pathlib import Path
from typing import Callable, Optional
from threading import Thread, Event
logger = logging.getLogger(__name__)
class ConfigWatcher:
"""配置文件监听器：文件变化时自动重载"""
def __init__(self, config_path: str, callback: Callable, check_interval: float = 5.0):
self.config_path = Path(config_path)
self.callback = callback
self.check_interval = check_interval
self._stop_event = Event()
self._last_hash = self._file_hash()
self._thread: Optional[Thread] = None
def _file_hash(self) -> str:
"""计算文件内容哈希"""
if not self.config_path.exists():
return ""
with open(self.config_path, "rb") as f:
return hashlib.md5(f.read()).hexdigest()
def _watch_loop(self):
"""监听循环"""
logger.info(f"开始监听配置文件: {self.config_path}")
while not self._stop_event.is_set():
try:
current_hash = self._file_hash()
if current_hash and current_hash != self._last_hash:
logger.info("检测到配置文件变更，重新加载...")
try:
self.callback()
self._last_hash = current_hash
logger.info("配置重载成功")
except Exception as e:
logger.error(f"配置重载失败: {e}")
except Exception as e:
logger.error(f"配置监听异常: {e}")
self._stop_event.wait(self.check_interval)
def start(self):
"""启动监听"""
self._stop_event.clear()
self._thread = Thread(target=self._watch_loop, daemon=True)
self._thread.start()
def stop(self):
"""停止监听"""
self._stop_event.set()
if self._thread:
self._thread.join(timeout=5)
# 使用
class AppConfig:
def __init__(self):
self.debug = False
self.log_level = "INFO"
self.max_workers = 4
def reload(self):
"""重载配置"""
from dotenv import load_dotenv
load_dotenv("config.env", override=True)
self.debug = os.getenv("DEBUG", "false").lower() in ("true", "1")
self.log_level = os.getenv("LOG_LEVEL", "INFO")
self.max_workers = int(os.getenv("MAX_WORKERS", "4"))
app_config = AppConfig()
watcher = ConfigWatcher("config.env", app_config.reload, check_interval=5.0)
watcher.start()
# 配置文件变化时自动调用 app_config.reload()Feature Flag 配置管理
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
from datetime import datetime, date
import json
@dataclass
class FeatureFlag:
"""功能开关"""
key: str
enabled: bool = False
description: str = ""
rollout_percent: int = 100 # 灰度百分比 0-100
whitelist: list[str] = field(default_factory=list) # 白名单用户 ID
start_date: Optional[str] = None # 生效日期
end_date: Optional[str] = None # 失效日期
metadata: Dict[str, Any] = field(default_factory=dict)
class FeatureFlagManager:
"""功能开关管理器"""
def __init__(self):
self._flags: Dict[str, FeatureFlag] = {}
self._user_flags: Dict[str, Dict[str, bool]] = {} # 用户级覆盖
def load_from_file(self, filepath: str):
"""从 JSON 文件加载配置"""
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
for key, config in data.get("flags", {}).items():
self._flags[key] = FeatureFlag(
key=key,
enabled=config.get("enabled", False),
description=config.get("description", ""),
rollout_percent=config.get("rollout_percent", 100),
whitelist=config.get("whitelist", []),
start_date=config.get("start_date"),
end_date=config.get("end_date"),
metadata=config.get("metadata", {}),
)
def load_from_env(self):
"""从环境变量加载覆盖"""
import os
for key in self._flags:
env_val = os.getenv(f"FF_{key.upper()}")
if env_val is not None:
self._flags[key].enabled = env_val.lower() in ("true", "1", "yes")
def is_enabled(self, key: str, user_id: Optional[str] = None) -> bool:
"""检查功能是否对指定用户启用"""
flag = self._flags.get(key)
if not flag:
return False
# 检查时间窗口
today = date.today().isoformat()
if flag.start_date and today < flag.start_date:
return False
if flag.end_date and today > flag.end_date:
return False
# 用户级覆盖
if user_id and user_id in self._user_flags:
return self._user_flags[user_id].get(key, flag.enabled)
# 白名单检查
if user_id and user_id in flag.whitelist:
return True
# 灰度检查（基于用户 ID 哈希）
if flag.enabled and user_id and flag.rollout_percent < 100:
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
return hash_val < flag.rollout_percent
return flag.enabled
def get_all_flags(self) -> Dict[str, dict]:
"""获取所有功能开关状态"""
return {
key: {
"enabled": flag.enabled,
"description": flag.description,
"rollout_percent": flag.rollout_percent,
}
for key, flag in self._flags.items()
}
# 使用
import hashlib
ff_manager = FeatureFlagManager()
# features.json:
# {
# "flags": {
# "new_ui": {"enabled": true, "description": "新版界面", "rollout_percent": 50},
# "dark_mode": {"enabled": true, "description": "暗色模式"},
# "beta_api": {"enabled": false, "description": "测试 API", "whitelist": ["U-001", "U-002"]}
# }
# }
# ff_manager.load_from_file("features.json")
# ff_manager.load_from_env()
# if ff_manager.is_enabled("new_ui", user_id="U-12345"):
# return render_new_ui()Kubernetes ConfigMap 与 Secret 集成
import os
import json
from pathlib import Path
from typing import Any, Dict
class K8sConfigLoader:
"""Kubernetes 配置加载器"""
# K8s ConfigMap 挂载路径（默认）
CONFIGMAP_PATH = "/etc/config"
# K8s Secret 挂载路径（默认）
SECRET_PATH = "/etc/secrets"
@classmethod
def load_configmap(cls, configmap_path: str = None) -> Dict[str, str]:
"""加载 ConfigMap 中的配置"""
base_path = Path(configmap_path or cls.CONFIGMAP_PATH)
config = {}
if not base_path.exists():
return config
for file_path in base_path.iterdir():
if file_path.is_file():
key = file_path.name
value = file_path.read_text(encoding="utf-8").strip()
config[key] = value
return config
@classmethod
def load_secrets(cls, secret_path: str = None) -> Dict[str, str]:
"""加载 Secret 中的配置"""
base_path = Path(secret_path or cls.SECRET_PATH)
secrets = {}
if not base_path.exists():
return secrets
for file_path in base_path.iterdir():
if file_path.is_file():
key = file_path.name
value = file_path.read_text(encoding="utf-8").strip()
secrets[key] = value
return secrets
@classmethod
def merge_configs(cls) -> Dict[str, Any]:
"""合并 ConfigMap、Secret 和环境变量（优先级递增）"""
config = {}
# 1. ConfigMap（最低优先级）
config.update(cls.load_configmap())
# 2. Secret
config.update(cls.load_secrets())
# 3. 环境变量（最高优先级）
for key, value in os.environ.items():
config[key] = value
return config
# 使用
# 在 K8s Pod 中，ConfigMap 和 Secret 会被挂载为文件
# k8s_config = K8sConfigLoader.merge_configs()
# db_password = k8s_config.get("DB_PASSWORD")配置管理最佳实践总结
# 1. 配置分层原则（优先级从低到高）
# 默认值（代码中） < 配置文件（YAML/TOML） < 环境变量 < 命令行参数
#
# 2. 敏感信息保护
# - 密码、Token、API Key 永远不写入代码仓库
# - 使用 .env 文件管理本地开发配置（.env.example 提交到仓库）
# - 生产环境通过环境变量或密钥管理服务注入
# - 使用 pydantic.SecretStr 自动脱敏日志输出
#
# 3. 配置变更管理
# - 配置文件纳入版本管理（除含敏感信息的文件）
# - 配置变更必须有审计日志
# - 生产环境配置变更需经过审批流程
# - 使用 Feature Flag 控制新功能的灰度发布
#
# 4. 配置校验清单
def startup_config_check(config) -> list[str]:
"""启动时配置检查清单"""
errors = []
# 必须项检查
required_keys = ["DATABASE_URL", "SECRET_KEY", "REDIS_URL"]
for key in required_keys:
if not getattr(config, key.lower(), None):
errors.append(f"缺少必须配置: {key}")
# 格式检查
db_url = getattr(config, "database_url", "")
if db_url and not db_url.startswith(("postgresql://", "mysql://", "sqlite://")):
errors.append(f"DATABASE_URL 格式不正确: {db_url[:20]}...")
# 范围检查
workers = getattr(config, "max_workers", 0)
if not (1 <= workers <= 64):
errors.append(f"max_workers 超出合理范围: {workers}")
# 安全检查
secret = getattr(config, "secret_key", "")
if secret and len(secret) < 32:
errors.append("SECRET_KEY 长度不足（建议 32+ 字符）")
debug = getattr(config, "debug", False)
env = getattr(config, "environment", "dev")
if debug and env == "prod":
errors.append("生产环境禁止开启 DEBUG 模式")
return errors
# 5. 多环境配置文件组织
# config/
# base.yaml # 基础配置（所有环境共享）
# development.yaml # 开发环境
# staging.yaml # 预发布环境
# production.yaml # 生产环境
# testing.yaml # 测试环境
# .env.example # 环境变量模板（提交到仓库）
# .env # 本地环境变量（不提交）优点
缺点
总结
配置管理的核心是建立清晰的配置层级：默认值 < 配置文件 < 环境变量 < 启动参数，确保每个环境的配置独立可控。推荐使用 pydantic-settings 作为主方案，它提供类型校验、环境变量绑定和 .env 文件支持。生产环境务必在启动时校验配置完整性，防止因缺失配置导致运行时错误。
关键知识点
- 环境变量是配置管理的基石，所有敏感信息必须通过环境变量注入
- pydantic-settings 自动将环境变量映射到配置类字段，支持类型校验
- 配置文件（YAML/TOML）适合非敏感的结构化配置，环境变量适合部署相关的值
- 十二因素应用（12-Factor App）原则要求严格分离代码和配置
项目落地视角
- 使用 .env 文件管理本地开发配置，.env.example 提交到仓库作为模板
- CI/CD 流水线通过环境变量注入生产配置，不在构建产物中包含敏感信息
- 服务启动时执行配置校验，缺失关键配置项时立即报错退出
- 为配置变更添加日志记录，便于排查因配置变更导致的问题
常见误区
- 将数据库密码、API Key 等敏感信息直接写在配置文件中并提交到代码仓库
- 硬编码配置值在代码中，导致每次修改都需要重新构建部署
- 忽略配置的默认值设计，导致新环境部署时因缺少配置而失败
- 配置文件没有版本管理和变更记录
进阶路线
- 学习分布式配置中心（Consul、etcd、Nacos）的集成方式
- 研究配置热更新机制，实现不停机更新配置
- 了解 Kubernetes ConfigMap 和 Secret 的使用方式
- 探索 Feature Flag 系统的配置管理模式
适用场景
- 多环境部署（开发/测试/预发布/生产）需要差异化配置的项目
- 需要严格保护敏感信息（数据库密码、API Key）的合规场景
- 微服务架构下需要统一配置管理和服务发现
落地建议
- 使用 pydantic-settings + .env 作为标准配置方案，团队统一
- .env.example 文件列出所有配置项及说明，新人可直接复制使用
- 配置校验集成到应用启动流程，失败时打印清晰的错误信息
排错清单
- 检查 .env 文件是否存在且编码正确（UTF-8，无 BOM）
- 确认环境变量名称和前缀是否与配置类定义匹配
- 排查多层配置覆盖的实际生效值，使用调试模式打印最终配置
复盘问题
- 你的项目中敏感信息是否全部通过环境变量管理？代码仓库中是否存在泄漏？
- 配置变更时是否有审计日志？能否快速回滚到之前的配置？
- 新环境部署时配置是否自动化？是否需要手动修改配置文件？