Python配置管理与环境变量一、环境变量基础import os# 读取环境变量db_host os.environ.get(DB_HOST, localhost)db_port int(os.environ.get(DB_PORT, 5432))debug os.environ.get(DEBUG, false).lower() in (true, 1, yes)# 必需的环境变量def require_env(name):value os.environ.get(name)if value is None:raise EnvironmentError(f缺少必需的环境变量: {name})return valueSECRET_KEY require_env(SECRET_KEY)DATABASE_URL require_env(DATABASE_URL)二、.env文件管理使用python-dotenv加载.env文件# .env文件内容# DB_HOSTlocalhost# DB_PORT5432# DB_NAMEmyapp# SECRET_KEYyour-secret-key-here# DEBUGtruefrom dotenv import load_dotenvfrom pathlib import Path# 加载.env文件env_path Path(.) / .envload_dotenv(dotenv_pathenv_path)# 或自动查找load_dotenv() # 自动查找当前目录及父目录的.env# 不覆盖已有环境变量load_dotenv(overrideFalse)三、分层配置类from dataclasses import dataclass, fieldfrom typing import Optionalimport osdataclassclass DatabaseConfig:host: str localhostport: int 5432name: str myappuser: str postgrespassword: str pool_size: int 5pool_timeout: int 30propertydef url(self) - str:return fpostgresql://{self.user}:{self.password}{self.host}:{self.port}/{self.name}dataclassclass RedisConfig:host: str localhostport: int 6379db: int 0password: Optional[str] Nonepropertydef url(self) - str:auth f:{self.password} if self.password else return fredis://{auth}{self.host}:{self.port}/{self.db}dataclassclass AppConfig:debug: bool Falsesecret_key: str host: str 0.0.0.0port: int 8000workers: int 4log_level: str INFOallowed_origins: list field(default_factorylambda: [*])database: DatabaseConfig field(default_factoryDatabaseConfig)redis: RedisConfig field(default_factoryRedisConfig)classmethoddef from_env(cls) - AppConfig:从环境变量创建配置return cls(debugos.environ.get(DEBUG, false).lower() true,secret_keyos.environ.get(SECRET_KEY, dev-secret),hostos.environ.get(HOST, 0.0.0.0),portint(os.environ.get(PORT, 8000)),workersint(os.environ.get(WORKERS, 4)),log_levelos.environ.get(LOG_LEVEL, INFO),databaseDatabaseConfig(hostos.environ.get(DB_HOST, localhost),portint(os.environ.get(DB_PORT, 5432)),nameos.environ.get(DB_NAME, myapp),useros.environ.get(DB_USER, postgres),passwordos.environ.get(DB_PASSWORD, ),),redisRedisConfig(hostos.environ.get(REDIS_HOST, localhost),portint(os.environ.get(REDIS_PORT, 6379)),),)四、Pydantic Settingsfrom pydantic_settings import BaseSettingsfrom pydantic import Field, validatorclass Settings(BaseSettings):使用Pydantic自动从环境变量加载配置# 基本配置app_name: str MyAppdebug: bool Falsesecret_key: str Field(..., min_length16)# 数据库db_host: str localhostdb_port: int 5432db_name: str myappdb_user: str postgresdb_password: str # Redisredis_url: str redis://localhost:6379/0# 服务配置host: str 0.0.0.0port: int 8000workers: int 4cors_origins: list[str] [*]validator(workers)def validate_workers(cls, v):if v 1 or v 32:raise ValueError(workers必须在1-32之间)return vpropertydef database_url(self) - str:return fpostgresql://{self.db_user}:{self.db_password}{self.db_host}:{self.db_port}/{self.db_name}class Config:env_file .envenv_file_encoding utf-8case_sensitive False# 环境变量前缀env_prefix # 使用settings Settings()print(settings.database_url)print(settings.debug)五、多环境配置import osfrom pathlib import Pathclass ConfigManager:多环境配置管理def __init__(self, envNone):self.env env or os.environ.get(APP_ENV, development)self._config {}self._load()def _load(self):按优先级加载配置# 1. 默认配置self._config self._load_defaults()# 2. 环境特定配置env_config self._load_env_config()self._deep_merge(self._config, env_config)# 3. 本地覆盖不提交到版本控制local_config self._load_local_config()self._deep_merge(self._config, local_config)# 4. 环境变量覆盖最高优先级self._apply_env_vars()def _load_defaults(self):return {app: {name: MyApp, debug: False, port: 8000},database: {host: localhost, port: 5432},logging: {level: INFO, format: json},}def _load_env_config(self):加载环境特定配置文件config_file Path(fconfig/{self.env}.yaml)if config_file.exists():import yamlwith open(config_file) as f:return yaml.safe_load(f) or {}return {}def _load_local_config(self):config_file Path(config/local.yaml)if config_file.exists():import yamlwith open(config_file) as f:return yaml.safe_load(f) or {}return {}def _apply_env_vars(self):环境变量覆盖格式APP__DATABASE__HOSTprefix APP__for key, value in os.environ.items():if key.startswith(prefix):parts key[len(prefix):].lower().split(__)self._set_nested(self._config, parts, value)def _set_nested(self, d, keys, value):for key in keys[:-1]:d d.setdefault(key, {})# 尝试类型转换d[keys[-1]] self._convert_value(value)def _convert_value(self, value):if value.lower() in (true, false):return value.lower() truetry:return int(value)except ValueError:try:return float(value)except ValueError:return valuedef _deep_merge(self, base, override):for key, value in override.items():if key in base and isinstance(base[key], dict) and isinstance(value, dict):self._deep_merge(base[key], value)else:base[key] valuedef get(self, key, defaultNone):支持点号分隔的键路径keys key.split(.)value self._configfor k in keys:if isinstance(value, dict):value value.get(k)else:return defaultif value is None:return defaultreturn value# 使用config ConfigManager() # 自动检测环境print(config.get(database.host))print(config.get(app.debug))六、配置验证from dataclasses import dataclassfrom typing import Anyclass ConfigValidator:配置验证器def __init__(self):self.rules []def require(self, key, type_str, validatorNone):self.rules.append({key: key,required: True,type: type_,validator: validator,})return selfdef optional(self, key, type_str, defaultNone, validatorNone):self.rules.append({key: key,required: False,type: type_,default: default,validator: validator,})return selfdef validate(self, config: dict) - dict:errors []validated {}for rule in self.rules:key rule[key]value self._get_nested(config, key)if value is None:if rule[required]:errors.append(f缺少必需配置: {key})continueelse:value rule.get(default)if value is not None:# 类型检查try:value rule[type](value)except (ValueError, TypeError) as e:errors.append(f{key}: 类型转换失败 ({e}))continue# 自定义验证if rule.get(validator):try:rule[validator](value)except ValueError as e:errors.append(f{key}: {e})continuevalidated[key] valueif errors:raise ValueError(f配置验证失败:\n \n.join(f - {e} for e in errors))return validateddef _get_nested(self, d, key):for k in key.split(.):if isinstance(d, dict):d d.get(k)else:return Nonereturn d# 使用validator ConfigValidator()validator.require(database.host, str)validator.require(database.port, int, lambda v: v if 1 v 65535 else (_ for _ in ()).throw(ValueError(端口范围1-65535)))validator.require(secret_key, str, lambda v: v if len(v) 16 else (_ for _ in ()).throw(ValueError(至少16个字符)))validator.optional(debug, bool, defaultFalse)validator.optional(workers, int, default4)七、敏感配置管理import base64from cryptography.fernet import Fernetclass SecretManager:敏感配置加密管理def __init__(self, master_keyNone):self.master_key master_key or os.environ.get(MASTER_KEY)if not self.master_key:raise ValueError(需要MASTER_KEY环境变量)self.cipher Fernet(self.master_key.encode())def encrypt(self, value: str) - str:return self.cipher.encrypt(value.encode()).decode()def decrypt(self, encrypted: str) - str:return self.cipher.decrypt(encrypted.encode()).decode()def encrypt_file(self, input_path, output_path):加密配置文件with open(input_path, r) as f:content f.read()encrypted self.encrypt(content)with open(output_path, w) as f:f.write(encrypted)def decrypt_file(self, input_path) - str:解密配置文件with open(input_path, r) as f:encrypted f.read()return self.decrypt(encrypted)staticmethoddef generate_key() - str:生成新的主密钥return Fernet.generate_key().decode()八、配置热重载import threadingimport timefrom pathlib import Pathclass HotReloadConfig:支持热重载的配置def __init__(self, config_path, reload_interval5):self.config_path Path(config_path)self.reload_interval reload_intervalself._config {}self._last_modified 0self._lock threading.Lock()self._callbacks []self._load()self._start_watcher()def _load(self):import yamlwith open(self.config_path) as f:new_config yaml.safe_load(f)with self._lock:old_config self._config.copy()self._config new_configself._last_modified self.config_path.stat().st_mtime# 通知变更if old_config and old_config ! new_config:for callback in self._callbacks:callback(old_config, new_config)def _start_watcher(self):def watch():while True:time.sleep(self.reload_interval)try:mtime self.config_path.stat().st_mtimeif mtime self._last_modified:self._load()print(f配置已重新加载: {self.config_path})except Exception as e:print(f重载配置失败: {e})thread threading.Thread(targetwatch, daemonTrue)thread.start()def on_change(self, callback):self._callbacks.append(callback)def get(self, key, defaultNone):with self._lock:keys key.split(.)value self._configfor k in keys:if isinstance(value, dict):value value.get(k)else:return defaultreturn value if value is not None else default# 使用config HotReloadConfig(config/app.yaml)config.on_change(lambda old, new: print(f配置变更: {old} - {new}))九、Feature Flagsclass FeatureFlags:功能开关管理def __init__(self, config_source):self._flags {}self._source config_sourceself._load()def _load(self):self._flags self._source.get(feature_flags, {})def is_enabled(self, flag_name, defaultFalse):flag self._flags.get(flag_name)if flag is None:return defaultif isinstance(flag, bool):return flagif isinstance(flag, dict):return flag.get(enabled, default)return defaultdef get_variant(self, flag_name, user_idNone):flag self._flags.get(flag_name, {})if not isinstance(flag, dict):return None# 百分比灰度percentage flag.get(percentage, 100)if user_id:hash_val hash(f{flag_name}:{user_id}) % 100if hash_val percentage:return Nonereturn flag.get(variant, default)# 配置示例# feature_flags:# new_checkout:# enabled: true# percentage: 50# variant: v2# dark_mode:# enabled: true十、总结配置管理最佳实践1. 敏感信息用环境变量不要提交到代码仓库2. 使用.env.example记录需要的环境变量3. 配置分层默认值 配置文件 环境变量4. 启动时验证配置完整性5. 使用Pydantic Settings获得类型安全6. 不同环境使用不同配置文件7. 支持配置热重载减少重启8. 加密存储敏感配置