149 lines
4.0 KiB
Python
149 lines
4.0 KiB
Python
"""
|
|
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
|
|
|