""" 配置加载器模块 提供统一的配置加载机制，支持多数据源配置合并： - 环境变量（优先级最高） - YAML 配置文件 - 默认值（优先级最低） 设计原则： - 单一职责：仅负责配置的加载和合并 - 依赖倒置：通过抽象接口解耦配置源 - 开闭原则：易于扩展新的配置源 """ from __future__ import annotations import os from abc import ABC, abstractmethod from functools import lru_cache from pathlib import Path from typing import Any, TypeVar import yaml T = TypeVar("T") class ConfigSourceError(Exception): """配置源加载错误""" def __init__(self, source: str, message: str) -> None: self.source = source self.message = message super().__init__(f"[{source}] {message}") class ConfigSource(ABC): """配置源抽象基类""" @property @abstractmethod def name(self) -> str: """配置源名称""" ... @property @abstractmethod def priority(self) -> int: """优先级，数值越大优先级越高""" ... @abstractmethod def load(self) -> dict[str, Any]: """加载配置，返回扁平化的键值对""" ... class EnvConfigSource(ConfigSource): """环境变量配置源""" def __init__(self, prefix: str = "") -> None: """ 初始化环境变量配置源 Args: prefix: 环境变量前缀，用于过滤相关配置 例如 prefix="SATONANO_" 时只读取以此为前缀的变量 """ self._prefix = prefix.upper() @property def name(self) -> str: return "environment" @property def priority(self) -> int: return 100 # 最高优先级 def load(self) -> dict[str, Any]: """ 加载环境变量 Returns: 配置字典，键名转换为小写并移除前缀 """ config: dict[str, Any] = {} prefix_len = len(self._prefix) for key, value in os.environ.items(): if self._prefix and not key.upper().startswith(self._prefix): continue # 移除前缀并转为小写 config_key = key[prefix_len:].lower() if self._prefix else key.lower() config[config_key] = self._parse_value(value) return config @staticmethod def _parse_value(value: str) -> Any: """ 解析环境变量值，自动转换类型 Args: value: 原始字符串值 Returns: 转换后的值（bool/int/float/str） """ # 布尔值 if value.lower() in ("true", "yes", "1", "on"): return True if value.lower() in ("false", "no", "0", "off"): return False # 整数 try: return int(value) except ValueError: pass # 浮点数 try: return float(value) except ValueError: pass return value class YamlConfigSource(ConfigSource): """YAML 配置文件源""" def __init__( self, file_path: str | Path = "config.yaml", required: bool = False, ) -> None: """ 初始化 YAML 配置源 Args: file_path: 配置文件路径 required: 是否必须存在，为 True 时文件不存在会抛出异常 """ self._file_path = Path(file_path) self._required = required @property def name(self) -> str: return f"yaml:{self._file_path}" @property def priority(self) -> int: return 50 # 中等优先级 def load(self) -> dict[str, Any]: """ 加载 YAML 配置文件 Returns: 配置字典 Raises: ConfigSourceError: 文件不存在（required=True）或解析错误 """ if not self._file_path.exists(): if self._required: raise ConfigSourceError( self.name, f"配置文件不存在: {self._file_path.absolute()}", ) return {} try: with open(self._file_path, encoding="utf-8") as f: data = yaml.safe_load(f) except yaml.YAMLError as e: raise ConfigSourceError(self.name, f"YAML 解析错误: {e}") from e if data is None: return {} # 处理列表格式（兼容用户示例格式） if isinstance(data, list): return self._flatten_list(data) # 标准字典格式 if isinstance(data, dict): return self._normalize_keys(data) raise ConfigSourceError( self.name, f"不支持的配置格式，期望 dict 或 list，得到 {type(data).__name__}", ) @staticmethod def _flatten_list(items: list) -> dict[str, Any]: """ 将列表格式配置展平为字典 支持格式: - key: value - key: value """ result: dict[str, Any] = {} for item in items: if isinstance(item, dict): for key, value in item.items(): result[key.lower()] = value return result @staticmethod def _normalize_keys(data: dict) -> dict[str, Any]: """键名标准化为小写""" return {k.lower(): v for k, v in data.items()} class ConfigLoader: """ 配置加载器 负责从多个配置源加载并合并配置，按优先级覆盖。 Example: >>> loader = ConfigLoader() >>> loader.add_source(YamlConfigSource("config.yaml")) >>> loader.add_source(EnvConfigSource("SATONANO_")) >>> config = loader.load() >>> db_type = loader.get("database_type", "sqlite") """ def __init__(self) -> None: self._sources: list[ConfigSource] = [] self._config: dict[str, Any] = {} self._loaded = False def add_source(self, source: ConfigSource) -> ConfigLoader: """ 添加配置源 Args: source: 配置源实例 Returns: self，支持链式调用 """ self._sources.append(source) self._loaded = False # 标记需要重新加载 return self def load(self) -> dict[str, Any]: """ 加载并合并所有配置源 Returns: 合并后的配置字典 """ # 按优先级升序排序，后加载的覆盖先加载的 sorted_sources = sorted(self._sources, key=lambda s: s.priority) self._config = {} for source in sorted_sources: source_config = source.load() self._config.update(source_config) self._loaded = True return self._config.copy() def get(self, key: str, default: T = None) -> T | Any: """ 获取配置值 Args: key: 配置键名（不区分大小写） default: 默认值 Returns: 配置值或默认值 """ if not self._loaded: self.load() return self._config.get(key.lower(), default) def get_str(self, key: str, default: str = "") -> str: """获取字符串配置""" value = self.get(key, default) return str(value) if value is not None else default def get_int(self, key: str, default: int = 0) -> int: """获取整数配置""" value = self.get(key, default) if isinstance(value, int): return value try: return int(value) except (ValueError, TypeError): return default def get_float(self, key: str, default: float = 0.0) -> float: """获取浮点数配置""" value = self.get(key, default) if isinstance(value, float): return value try: return float(value) except (ValueError, TypeError): return default def get_bool(self, key: str, default: bool = False) -> bool: """获取布尔配置""" value = self.get(key, default) if isinstance(value, bool): return value if isinstance(value, str): return value.lower() in ("true", "yes", "1", "on") return bool(value) def require(self, key: str) -> Any: """ 获取必需的配置值 Args: key: 配置键名 Returns: 配置值 Raises: KeyError: 配置不存在 """ if not self._loaded: self.load() key_lower = key.lower() if key_lower not in self._config: raise KeyError(f"缺少必需的配置项: {key}") return self._config[key_lower] def all(self) -> dict[str, Any]: """获取所有配置的副本""" if not self._loaded: self.load() return self._config.copy() def __contains__(self, key: str) -> bool: """检查配置是否存在""" if not self._loaded: self.load() return key.lower() in self._config def __repr__(self) -> str: sources = ", ".join(s.name for s in self._sources) return f"ConfigLoader(sources=[{sources}], loaded={self._loaded})" def create_config_loader( yaml_path: str | Path = "config.yaml", env_prefix: str = "SATONANO_", yaml_required: bool = False, ) -> ConfigLoader: """ 创建预配置的配置加载器 Args: yaml_path: YAML 配置文件路径 env_prefix: 环境变量前缀 yaml_required: YAML 文件是否必须存在 Returns: 配置好的 ConfigLoader 实例 """ loader = ConfigLoader() loader.add_source(YamlConfigSource(yaml_path, required=yaml_required)) loader.add_source(EnvConfigSource(env_prefix)) return loader @lru_cache def get_config_loader() -> ConfigLoader: """获取全局配置加载器单例""" return create_config_loader() # 便捷访问的全局实例 config_loader = get_config_loader()