""" FastAPI 依赖注入模块 本模块提供 FastAPI 的依赖注入函数,用于在路由中注入可复用的依赖项。 使用依赖注入可以减少代码重复,提高可测试性和可维护性。 提供的依赖项: - get_logger: 获取日志记录器 - get_app_tools: 获取应用工具实例 - get_f6_module: 获取 F6Module 实例 - get_f6_plugin_module: 获取 F6PluginModule 实例 - get_other_module: 获取 OtherPluginModule 实例 - get_action_map: 获取操作映射表 """ from fastapi import Request from typing import Optional import logging def get_logger(request: Request) -> logging.Logger: """ 获取日志记录器依赖项 从应用状态中获取日志记录器,如果未初始化则返回一个基本的 logger。 Args: request: FastAPI 请求对象 Returns: logging.Logger: 日志记录器实例 """ logger = getattr(request.app.state, 'logger', None) if logger is None: # 如果 logger 未初始化,返回一个基本的 logger logger = logging.getLogger('app') return logger def get_app_tools(request: Request): """ 获取应用工具实例依赖项 从应用状态中获取 AppTools 实例,用于任务队列管理等操作。 Args: request: FastAPI 请求对象 Returns: AppTools: 应用工具实例 Raises: RuntimeError: 如果 AppTools 未初始化 """ from app.utils.app_tools import AppTools app_tools = getattr(request.app.state, 'app_tools', None) if app_tools is None: raise RuntimeError("AppTools 未初始化") return app_tools def get_f6_module(request: Request): """ 获取 F6Module 实例依赖项 从应用状态中获取 F6Module 实例,用于 F6 系统相关操作。 Args: request: FastAPI 请求对象 Returns: F6Module: F6Module 实例 Raises: RuntimeError: 如果 F6Module 未初始化 """ f6_module = getattr(request.app.state, 'f6_module', None) if f6_module is None: raise RuntimeError("F6Module 未初始化") return f6_module def get_f6_plugin_module(request: Request): """ 获取 F6PluginModule 实例依赖项 从应用状态中获取 F6PluginModule 实例,用于 F6 插件相关操作。 Args: request: FastAPI 请求对象 Returns: F6PluginModule: F6PluginModule 实例 Raises: RuntimeError: 如果 F6PluginModule 未初始化 """ f6_plugin_module = getattr(request.app.state, 'f6_plugin_module', None) if f6_plugin_module is None: raise RuntimeError("F6PluginModule 未初始化") return f6_plugin_module def get_other_module(request: Request): """ 获取 OtherPluginModule 实例依赖项 从应用状态中获取 OtherPluginModule 实例,用于其他插件相关操作。 Args: request: FastAPI 请求对象 Returns: OtherPluginModule: OtherPluginModule 实例 Raises: RuntimeError: 如果 OtherPluginModule 未初始化 """ other_module = getattr(request.app.state, 'other_module', None) if other_module is None: raise RuntimeError("OtherPluginModule 未初始化") return other_module def get_action_map(request: Request) -> dict: """ 获取操作映射表依赖项 从模块注册表中获取所有注册的操作,并转换为字典格式(操作名 -> 处理函数)。 Args: request: FastAPI 请求对象 Returns: dict: 操作映射表,格式为 {操作名: 处理函数} """ from app.core.module_registry import registry # 从 registry 获取所有注册的操作 actions = registry.get_all_actions() # 转换为字典格式(handler 函数) action_map = { action_name: config.handler for action_name, config in actions.items() } return action_map