Python配置管理与环境变量

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 value

SECRET_KEY = require_env('SECRET_KEY')
DATABASE_URL = require_env('DATABASE_URL')


二、.env文件管理

使用python-dotenv加载.env文件：

# .env文件内容
# DB_HOST=localhost
# DB_PORT=5432
# DB_NAME=myapp
# SECRET_KEY=your-secret-key-here
# DEBUG=true

from dotenv import load_dotenv
from pathlib import Path

# 加载.env文件
env_path = Path('.') / '.env'
load_dotenv(dotenv_path=env_path)

# 或自动查找
load_dotenv() # 自动查找当前目录及父目录的.env

# 不覆盖已有环境变量
load_dotenv(override=False)


三、分层配置类

from dataclasses import dataclass, field
from typing import Optional
import os

@dataclass
class DatabaseConfig:
host: str = 'localhost'
port: int = 5432
name: str = 'myapp'
user: str = 'postgres'
password: str = ''
pool_size: int = 5
pool_timeout: int = 30

@property
def url(self) -> str:
return f"postgresql://{self.user}:{self.password}@{self.host}:{self.port}/{self.name}"

@dataclass
class RedisConfig:
host: str = 'localhost'
port: int = 6379
db: int = 0
password: Optional[str] = None

@property
def url(self) -> str:
auth = f":{self.password}@" if self.password else ""
return f"redis://{auth}{self.host}:{self.port}/{self.db}"

@dataclass
class AppConfig:
debug: bool = False
secret_key: str = ''
host: str = '0.0.0.0'
port: int = 8000
workers: int = 4
log_level: str = 'INFO'
allowed_origins: list = field(default_factory=lambda: ['*'])

database: DatabaseConfig = field(default_factory=DatabaseConfig)
redis: RedisConfig = field(default_factory=RedisConfig)

@classmethod
def from_env(cls) -> 'AppConfig':
"""从环境变量创建配置"""
return cls(
debug=os.environ.get('DEBUG', 'false').lower() == 'true',
secret_key=os.environ.get('SECRET_KEY', 'dev-secret'),
host=os.environ.get('HOST', '0.0.0.0'),
port=int(os.environ.get('PORT', '8000')),
workers=int(os.environ.get('WORKERS', '4')),
log_level=os.environ.get('LOG_LEVEL', 'INFO'),
database=DatabaseConfig(
host=os.environ.get('DB_HOST', 'localhost'),
port=int(os.environ.get('DB_PORT', '5432')),
name=os.environ.get('DB_NAME', 'myapp'),
user=os.environ.get('DB_USER', 'postgres'),
password=os.environ.get('DB_PASSWORD', ''),
),
redis=RedisConfig(
host=os.environ.get('REDIS_HOST', 'localhost'),
port=int(os.environ.get('REDIS_PORT', '6379')),
),
)


四、Pydantic Settings

from pydantic_settings import BaseSettings
from pydantic import Field, validator

class Settings(BaseSettings):
"""使用Pydantic自动从环境变量加载配置"""
# 基本配置
app_name: str = "MyApp"
debug: bool = False
secret_key: str = Field(..., min_length=16)

# 数据库
db_host: str = "localhost"
db_port: int = 5432
db_name: str = "myapp"
db_user: str = "postgres"
db_password: str = ""

# Redis
redis_url: str = "redis://localhost:6379/0"

# 服务配置
host: str = "0.0.0.0"
port: int = 8000
workers: int = 4
cors_origins: list[str] = ["*"]

@validator('workers')
def validate_workers(cls, v):
if v < 1 or v > 32:
raise ValueError("workers必须在1-32之间")
return v

@property
def database_url(self) -> str:
return f"postgresql://{self.db_user}:{self.db_password}@{self.db_host}:{self.db_port}/{self.db_name}"

class Config:
env_file = '.env'
env_file_encoding = 'utf-8'
case_sensitive = False
# 环境变量前缀
env_prefix = ''

# 使用
settings = Settings()
print(settings.database_url)
print(settings.debug)


五、多环境配置

import os
from pathlib import Path

class ConfigManager:
"""多环境配置管理"""
def __init__(self, env=None):
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(f'config/{self.env}.yaml')
if config_file.exists():
import yaml
with 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 yaml
with open(config_file) as f:
return yaml.safe_load(f) or {}
return {}

def _apply_env_vars(self):
"""环境变量覆盖（格式：APP__DATABASE__HOST）"""
prefix = '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() == 'true'
try:
return int(value)
except ValueError:
try:
return float(value)
except ValueError:
return value

def _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] = value

def get(self, key, default=None):
"""支持点号分隔的键路径"""
keys = key.split('.')
value = self._config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
else:
return default
if value is None:
return default
return value

# 使用
config = ConfigManager() # 自动检测环境
print(config.get('database.host'))
print(config.get('app.debug'))


六、配置验证

from dataclasses import dataclass
from typing import Any

class ConfigValidator:
"""配置验证器"""
def __init__(self):
self.rules = []

def require(self, key, type_=str, validator=None):
self.rules.append({
'key': key,
'required': True,
'type': type_,
'validator': validator,
})
return self

def optional(self, key, type_=str, default=None, validator=None):
self.rules.append({
'key': key,
'required': False,
'type': type_,
'default': default,
'validator': validator,
})
return self

def 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}")
continue
else:
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}")
continue

validated[key] = value

if errors:
raise ValueError(f"配置验证失败:\n" + "\n".join(f" - {e}" for e in errors))

return validated

def _get_nested(self, d, key):
for k in key.split('.'):
if isinstance(d, dict):
d = d.get(k)
else:
return None
return 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, default=False)
validator.optional('workers', int, default=4)


七、敏感配置管理

import base64
from cryptography.fernet import Fernet

class SecretManager:
"""敏感配置加密管理"""
def __init__(self, master_key=None):
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)

@staticmethod
def generate_key() -> str:
"""生成新的主密钥"""
return Fernet.generate_key().decode()


八、配置热重载

import threading
import time
from pathlib import Path

class HotReloadConfig:
"""支持热重载的配置"""
def __init__(self, config_path, reload_interval=5):
self.config_path = Path(config_path)
self.reload_interval = reload_interval
self._config = {}
self._last_modified = 0
self._lock = threading.Lock()
self._callbacks = []

self._load()
self._start_watcher()

def _load(self):
import yaml
with open(self.config_path) as f:
new_config = yaml.safe_load(f)

with self._lock:
old_config = self._config.copy()
self._config = new_config
self._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_mtime
if mtime > self._last_modified:
self._load()
print(f"配置已重新加载: {self.config_path}")
except Exception as e:
print(f"重载配置失败: {e}")

thread = threading.Thread(target=watch, daemon=True)
thread.start()

def on_change(self, callback):
self._callbacks.append(callback)

def get(self, key, default=None):
with self._lock:
keys = key.split('.')
value = self._config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
else:
return default
return value if value is not None else default

# 使用
config = HotReloadConfig('config/app.yaml')
config.on_change(lambda old, new: print(f"配置变更: {old} -> {new}"))


九、Feature Flags

class FeatureFlags:
"""功能开关管理"""
def __init__(self, config_source):
self._flags = {}
self._source = config_source
self._load()

def _load(self):
self._flags = self._source.get('feature_flags', {})

def is_enabled(self, flag_name, default=False):
flag = self._flags.get(flag_name)
if flag is None:
return default
if isinstance(flag, bool):
return flag
if isinstance(flag, dict):
return flag.get('enabled', default)
return default

def get_variant(self, flag_name, user_id=None):
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}") % 100
if hash_val >= percentage:
return None

return 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. 加密存储敏感配置