import sys from pathlib import Path from typing import TypedDict, Literal, Optional, cast, List, Dict, Any # 根据Python版本选择合适的TOML解析库和Required类型 if sys.version_info >= (3, 11): import tomllib from typing import Required else: import tomli as tomllib from typing_extensions import Required # ============================================================ # Worker 配置类型 # ============================================================ class WorkerParams(TypedDict, total=False): """Worker 操作参数定义""" type: str required: bool default: Any description: str class WorkerResponse(TypedDict, total=False): """Worker 操作响应定义""" type: str required: bool default: Any class OperationConfig(TypedDict, total=False): """ Worker 操作配置结构 对应 config/workers//*.toml 中的每个操作定义。 """ name: Required[str] method: Required[Literal["GET", "POST", "PUT", "DELETE"]] path: str subject: Required[str] summary: str timeout: float params: Dict[str, WorkerParams] response_model: Dict[str, WorkerResponse] class WorkerConfig(TypedDict, total=False): """ Worker 配置结构 对应 config/workers//config.toml 中的顶级配置。 """ name: Required[str] type: Required[str] description: str version: str enabled: bool operations: List[OperationConfig] def load_worker_config(config_dir: Path) -> Dict[str, Any]: """ 加载 Worker 配置目录中的所有 TOML 配置文件 Args: config_dir: Worker 配置目录路径,如 Path("config/workers/example") Returns: 解析后的配置字典,包含所有操作的配置 Raises: FileNotFoundError: 当配置目录不存在时抛出异常 """ if not config_dir.exists(): raise FileNotFoundError(f"Worker 配置目录不存在: {config_dir}") config: Dict[str, Any] = {"operations": []} # 首先加载 config.toml(如果存在) config_file = config_dir / "config.toml" if config_file.exists(): with open(config_file, "rb") as f: base_config = tomllib.load(f) config.update(base_config) # 然后加载所有 *.toml 中的操作定义 for toml_file in sorted(config_dir.glob("*.toml")): if toml_file.name == "config.toml": continue with open(toml_file, "rb") as f: op_data = tomllib.load(f) # 将文件名(不含扩展名)作为默认操作名 op_name = toml_file.stem if "name" not in op_data: op_data["name"] = op_name config["operations"].append(op_data) return config # ============================================================ # 旧的配置系统(与主应用 config.py 兼容) # ============================================================ # 设置 env 为 development 或 production class DefaultApp(TypedDict): """应用默认配置结构""" name: str version: str description: str debug: bool class Default(TypedDict): """系统默认配置结构,包含应用信息、环境、路径等基础配置""" app: DefaultApp env: Literal["production", "development"] prefix: str upload_path: str class SqliteConnectionArgs(TypedDict, total=False): """SQLite数据库连接参数配置""" timeout: int check_same_thread: bool class DatabaseSqlite(TypedDict, total=False): """SQLite数据库连接池及行为配置""" echo: bool pool_size: int max_overflow: int pool_timeout: int pool_recycle: int pool_pre_ping: bool connect_args: SqliteConnectionArgs class Database(TypedDict, total=False): """数据库配置结构,支持SQLite、MySQL、PostgreSQL三种数据库类型""" type: Required[Literal["sqlite", "mysql", "postgresql"]] url: Required[str] generate_schemas: bool sqlite: DatabaseSqlite mysql: dict postgresql: dict database_path: str class Security(TypedDict): """安全相关配置,包括JWT令牌和安全策略设置""" secret_key: str algorithm: Literal["HS256"] access_token_expire_minutes: int refresh_token_expire_days: int max_sessions_per_user: int class Server(TypedDict, total=False): """服务器运行配置,包含主机、端口、工作进程等设置""" host: Required[str] port: Required[int] reload: bool workers: int access_log: bool class Logging(TypedDict): """日志系统配置,定义日志级别、输出格式和文件管理策略""" level: str file: str format: str max_size: int backup_count: int class Cors(TypedDict, total=False): """跨域资源共享(CORS)配置""" allow_origins: List[str] allow_credentials: bool allow_methods: List[str] allow_headers: List[str] class ModuleSystem(TypedDict, total=False): """模块系统配置,定义模块加载类型和路径""" type: str path: List[str] class FileSystem(TypedDict, total=False): """文件系统配置,定义存储类型和路径""" type: str path: str class Env(TypedDict, total=False): """环境特定配置结构,用于开发和生产环境的差异化配置""" debug: Required[bool] server: Required[Server] logging: Required[Logging] cors: Cors module_system: ModuleSystem file_system: FileSystem class Scheduler(TypedDict): """任务调度器配置""" job_store_path: str class Cache(TypedDict): """缓存系统配置""" cache_store_path: str class Config(TypedDict): """完整配置结构,整合所有配置项,包含默认配置、数据库、调度器、缓存、安全和环境配置""" default: Default database: Database scheduler: Scheduler cache: Cache security: Security development: Env production: Env def load_config(config_path: Path = Path("config.toml")) -> Config: """ 从TOML配置文件加载配置数据 Args: config_path: 配置文件路径,默认为当前目录下的config.toml Returns: Config: 解析后的配置字典对象 Raises: FileNotFoundError: 当配置文件不存在时抛出异常 """ if not config_path.exists(): raise FileNotFoundError(f"配置文件不存在: {config_path}") with open(config_path, "rb") as f: return cast(Config, tomllib.load(f))