提供基本前后端骨架

2026-01-06 23:49:23 +08:00
parent 84d4ccc226
commit 06f8176e23
89 changed files with 19293 additions and 2 deletions
--- a/app/core/config_loader.py
+++ b/app/core/config_loader.py
@@ -0,0 +1,387 @@
+"""
+配置加载器模块
+
+提供统一的配置加载机制，支持多数据源配置合并：
+- 环境变量（优先级最高）
+- 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()
+