"""中心化配置模块 —— 读取项目根目录 env.yaml,Pydantic 校验导出强类型配置对象。""" from pathlib import Path from typing import Literal import yaml from pydantic import BaseModel, Field class DbConfig(BaseModel): """TimescaleDB / PostgreSQL 连接参数""" host: str = "localhost" port: int = Field(default=5432, ge=1, le=65535) name: str user: str password: str class RedisConfig(BaseModel): """Redis 连接配置""" url: str = "redis://localhost:6379" publish_enabled: bool = True class LoggingConfig(BaseModel): """日志配置""" level: Literal["trace", "debug", "info", "warn", "error", "fatal"] = "info" node_env: Literal["development", "production"] = "development" class EnvConfig(BaseModel): """env.yaml 顶层结构""" db: DbConfig redis: RedisConfig logging: LoggingConfig def _get_project_root() -> Path: return Path(__file__).resolve().parent.parent.parent def _load_yaml_config() -> dict: root = _get_project_root() yaml_path = root / "env.yaml" if not yaml_path.exists(): raise FileNotFoundError( f"[config] 无法读取配置文件: {yaml_path}\n" f"请确保项目根目录存在 env.yaml。" ) with open(yaml_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) if data is None: raise ValueError(f"[config] env.yaml 解析结果为空: {yaml_path}") return data _raw = _load_yaml_config() _config = EnvConfig.model_validate(_raw) db = _config.db redis = _config.redis logging = _config.logging DB_DSN = f"postgresql://{db.user}:{db.password}@{db.host}:{db.port}/{db.name}" def print_config_summary() -> None: """打印脱敏后的配置概要(不含密码明文)""" print(f"[config] TimescaleDB: {db.user}@{db.host}:{db.port}/{db.name}") print( f"[config] Redis: {redis.url.replace('//', '//***@') if '@' in redis.url else redis.url}" ) print(f"[config] Logging: level={logging.level}, env={logging.node_env}")