From 92f182cd093e536d53728ead66ddbeead98143b9 Mon Sep 17 00:00:00 2001 From: Toby Wong <1751001928@qq.com> Date: Fri, 17 Apr 2026 16:54:48 +0800 Subject: [PATCH] Add plugin depends on and lifecycle ordering (#1153) * feat: add priority support to LifespanManager.register * feat: support lifespan stages and plugin dependencies * Optimize implementations * Fix dependency resolution * Fix the depends on type --------- Co-authored-by: Wu Clan --- .schemas/plugin.schema.json | 9 ++ backend/app/admin/tests/utils/__init__.py | 0 backend/common/dataclasses.py | 10 ++ backend/common/enums.py | 8 ++ backend/common/lifespan.py | 49 ++++++-- backend/plugin/core.py | 144 +++++++++++++++------- backend/plugin/validator.py | 14 +++ 7 files changed, 178 insertions(+), 56 deletions(-) delete mode 100644 backend/app/admin/tests/utils/__init__.py diff --git a/.schemas/plugin.schema.json b/.schemas/plugin.schema.json index ac92ec59..4275fc81 100644 --- a/.schemas/plugin.schema.json +++ b/.schemas/plugin.schema.json @@ -61,6 +61,15 @@ "enum": ["mysql", "postgresql"] }, "x-tombi-array-values-order": "ascending" + }, + "depends_on": { + "type": "array", + "description": "List of plugin names this plugin depends on (used for startup order)", + "items": { + "type": "string", + "minLength": 1 + }, + "x-tombi-array-values-order": "ascending" } } }, diff --git a/backend/app/admin/tests/utils/__init__.py b/backend/app/admin/tests/utils/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/backend/common/dataclasses.py b/backend/common/dataclasses.py index b0dfdec2..a776f645 100644 --- a/backend/common/dataclasses.py +++ b/backend/common/dataclasses.py @@ -1,6 +1,7 @@ import dataclasses from datetime import datetime +from typing import Any from fastapi import Response @@ -73,3 +74,12 @@ class SnowflakeInfo: datacenter_id: int worker_id: int sequence: int + + +@dataclasses.dataclass(slots=True) +class PluginEntry: + name: str + depends_on: list[str] | None = None + extend: str | None = None + routers: list[str] | None = None + api: dict[str, Any] | None = None diff --git a/backend/common/enums.py b/backend/common/enums.py index 354e6aa1..38ac1e89 100644 --- a/backend/common/enums.py +++ b/backend/common/enums.py @@ -145,3 +145,11 @@ class PrimaryKeyType(StrEnum): autoincrement = 'autoincrement' snowflake = 'snowflake' + + +class LifespanStage(IntEnum): + """lifespan 执行阶段""" + + core = 0 + plugin = 1 + tail = 2 diff --git a/backend/common/lifespan.py b/backend/common/lifespan.py index 534f104f..ea82db56 100644 --- a/backend/common/lifespan.py +++ b/backend/common/lifespan.py @@ -1,9 +1,11 @@ from collections.abc import Callable from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager -from typing import Any +from typing import Any, overload from fastapi import FastAPI +from backend.common.enums import LifespanStage + LifespanFunc = Callable[[FastAPI], AbstractAsyncContextManager[dict[str, Any] | None]] @@ -11,18 +13,42 @@ class LifespanManager: """FastAPI lifespan 管理器""" def __init__(self) -> None: - self._lifespans: list[LifespanFunc] = [] + self._lifespans: dict[LifespanStage, list[LifespanFunc]] = { + LifespanStage.core: [], + LifespanStage.plugin: [], + LifespanStage.tail: [], + } - def register(self, func: LifespanFunc) -> LifespanFunc: + @overload + def register(self, func: LifespanFunc) -> LifespanFunc: ... + + @overload + def register(self, *, stage: LifespanStage) -> Callable[[LifespanFunc], LifespanFunc]: ... + + def register( + self, func: LifespanFunc | None = None, *, stage: LifespanStage = LifespanStage.core + ) -> LifespanFunc | Callable[[LifespanFunc], LifespanFunc]: """ 注册 lifespan hook - :param func: lifespan hook + :param func: lifespan hook（直接装饰时使用） + :param stage: 执行阶段，控制粗粒度顺序，默认为 core :return: """ - if func not in self._lifespans: - self._lifespans.append(func) - return func + + def decorator(f: LifespanFunc) -> LifespanFunc: + for hooks in self._lifespans.values(): + for fn in hooks: + if fn is f: + return f + + self._lifespans[stage].append(f) + return f + + if func is not None: + return decorator(func) + + return decorator def build(self) -> LifespanFunc: """ @@ -35,10 +61,11 @@ class LifespanManager: async def combined_lifespan(app: FastAPI): # noqa: ANN202 state: dict[str, Any] = {} async with AsyncExitStack() as exit_stack: - for lifespan_fn in self._lifespans: - result = await exit_stack.enter_async_context(lifespan_fn(app)) - if isinstance(result, dict): - state.update(result) + for stage in LifespanStage: + for lifespan_fn in self._lifespans[stage]: + result = await exit_stack.enter_async_context(lifespan_fn(app)) + if isinstance(result, dict): + state.update(result) for key, value in state.items(): setattr(app.state, key, value) diff --git a/backend/plugin/core.py b/backend/plugin/core.py index 132911d8..c0c170cb 100644 --- a/backend/plugin/core.py +++ b/backend/plugin/core.py @@ -11,7 +11,8 @@ import rtoml from fastapi import APIRouter, Depends, FastAPI, Request -from backend.common.enums import DataBaseType, PluginLevelType, PrimaryKeyType, StatusType +from backend.common.dataclasses import PluginEntry +from backend.common.enums import DataBaseType, LifespanStage, PluginLevelType, PrimaryKeyType, StatusType from backend.common.exception import errors from backend.common.lifespan import lifespan_manager from backend.common.log import log @@ -191,7 +192,7 @@ def register_plugin_lifespan_hook(plugin: str, module: Any) -> None: log.warning(f'插件 {plugin} 的 lifespan 不是可调用对象，已跳过') return - lifespan_manager.register(lifespan_hook) + lifespan_manager.register(lifespan_hook, stage=LifespanStage.plugin) # type: ignore[call-overload] log.info(f'插件 {plugin} lifespan hook 注册成功') @@ -218,11 +219,11 @@ def run_plugin_startup_hook(plugin: str, module: Any, app: FastAPI) -> None: log.info(f'插件 {plugin} startup hook 执行成功') -def parse_plugin_config() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: +def parse_plugin_config() -> tuple[list[PluginEntry], list[PluginEntry]]: """解析插件配置""" - extend_plugins = [] - app_plugins = [] plugins = get_plugins() + extend_plugins: list[PluginEntry] = [] + app_plugins: list[PluginEntry] = [] # 使用独立连接 current_redis_client = RedisCli() @@ -240,17 +241,25 @@ def parse_plugin_config() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: plugin_config = load_plugin_config(plugin) plugin_type = validate_plugin_config(plugin, plugin_config) - if plugin_type == PluginLevelType.extend: - extend_plugins.append(plugin_config) - else: - app_plugins.append(plugin_config) - # 补充插件信息 plugin_config['plugin']['name'] = plugin plugin_cache_key = f'{settings.PLUGIN_REDIS_PREFIX}:{plugin}' plugin_cache_info = run_await(current_redis_client.get)(plugin_cache_key) plugin_config['plugin']['enable'] = get_plugin_enable(plugin_cache_info, StatusType.enable.value) + plugin_entry = PluginEntry( + name=plugin, + depends_on=plugin_config['plugin'].get('depends_on'), + extend=plugin_config['app']['extend'] if plugin_type == PluginLevelType.extend else None, + routers=plugin_config['app']['router'] if plugin_type == PluginLevelType.app else None, + api=plugin_config['api'] if plugin_type == PluginLevelType.extend else None, + ) + + if plugin_type == PluginLevelType.extend: + extend_plugins.append(plugin_entry) + else: + app_plugins.append(plugin_entry) + # 缓存最新插件信息 run_await(current_redis_client.set)(plugin_cache_key, json.dumps(plugin_config, ensure_ascii=False)) @@ -262,17 +271,16 @@ def parse_plugin_config() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: return extend_plugins, app_plugins -def inject_extend_router(plugin: dict[str, Any]) -> None: +def inject_extend_router(plugin: PluginEntry) -> None: """ 扩展级插件路由注入 :param plugin: 插件名称 :return: """ - plugin_name: str = plugin['plugin']['name'] - plugin_api_path = PLUGIN_DIR / plugin_name / 'api' + plugin_api_path = PLUGIN_DIR / plugin.name / 'api' if not os.path.exists(plugin_api_path): - raise PluginConfigError(f'插件 {plugin} 缺少 api 目录，请检查插件文件是否完整') + raise PluginConfigError(f'插件 {plugin.name} 缺少 api 目录，请检查插件文件是否完整') for root, _, api_files in os.walk(plugin_api_path): for file in api_files: @@ -280,7 +288,7 @@ def inject_extend_router(plugin: dict[str, Any]) -> None: continue # 解析插件路由配置 - file_config = plugin['api'][file[:-3]] + file_config = plugin.api[file[:-3]] prefix = file_config['prefix'] tags = file_config['tags'] @@ -294,21 +302,21 @@ def inject_extend_router(plugin: dict[str, Any]) -> None: plugin_router = getattr(module, 'router', None) if not plugin_router: warnings.warn( - f'扩展级插件 {plugin_name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', + f'扩展级插件 {plugin.name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', FutureWarning, ) continue # 获取目标 app 路由 relative_path = os.path.relpath(root, plugin_api_path) - app_name = plugin.get('app', {}).get('extend') + app_name = plugin.extend target_module_path = f'backend.app.{app_name}.api.{relative_path.replace(os.sep, ".")}' target_module = import_module_cached(target_module_path) target_router = getattr(target_module, 'router', None) if not target_router or not isinstance(target_router, APIRouter): raise PluginInjectError( - f'扩展级插件 {plugin_name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', + f'扩展级插件 {plugin.name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', ) # 将插件路由注入到目标路由中 @@ -316,13 +324,13 @@ def inject_extend_router(plugin: dict[str, Any]) -> None: router=plugin_router, prefix=prefix, tags=[tags] if tags else [], - dependencies=[Depends(PluginStatusChecker(plugin_name))], + dependencies=[Depends(PluginStatusChecker(plugin.name))], ) except Exception as e: - raise PluginInjectError(f'扩展级插件 {plugin_name} 路由注入失败：{e!s}') from e + raise PluginInjectError(f'扩展级插件 {plugin.name} 路由注入失败：{e!s}') from e -def inject_app_router(plugin: dict[str, Any], target_router: APIRouter) -> None: +def inject_app_router(plugin: PluginEntry, target_router: APIRouter) -> None: """ 应用级插件路由注入 @@ -330,43 +338,84 @@ def inject_app_router(plugin: dict[str, Any], target_router: APIRouter) -> None: :param target_router: FastAPI 路由器 :return: """ - plugin_name: str = plugin['plugin']['name'] - module_path = f'backend.plugin.{plugin_name}.api.router' + module_path = f'backend.plugin.{plugin.name}.api.router' try: module = import_module_cached(module_path) - routers = plugin['app']['router'] + routers = plugin.routers if not routers or not isinstance(routers, list): - raise PluginConfigError(f'应用级插件 {plugin_name} 配置文件存在错误，请检查') + raise PluginConfigError(f'应用级插件 {plugin.name} 配置文件存在错误，请检查') for router in routers: plugin_router = getattr(module, router, None) if not plugin_router or not isinstance(plugin_router, APIRouter): raise PluginInjectError( - f'应用级插件 {plugin_name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', + f'应用级插件 {plugin.name} 模块 {module_path} 中没有有效的 router，请检查插件文件是否完整', ) # 将插件路由注入到目标路由中 - target_router.include_router(plugin_router, dependencies=[Depends(PluginStatusChecker(plugin_name))]) + target_router.include_router(plugin_router, dependencies=[Depends(PluginStatusChecker(plugin.name))]) except Exception as e: - raise PluginInjectError(f'应用级插件 {plugin_name} 路由注入失败：{e!s}') from e + raise PluginInjectError(f'应用级插件 {plugin.name} 路由注入失败：{e!s}') from e def build_final_router() -> APIRouter: """构建最终路由""" extend_plugins, app_plugins = parse_plugin_config() + plugins = extend_plugins + app_plugins + ordered_plugins = resolve_plugin_order(plugins) - for plugin in extend_plugins: - inject_extend_router(plugin) + for plugin in ordered_plugins: + if plugin.api is not None: + inject_extend_router(plugin) # 主路由，必须在扩展级插件路由注入后，应用级插件路由注入前导入 from backend.app.router import router as main_router - for plugin in app_plugins: - inject_app_router(plugin, main_router) + for plugin in ordered_plugins: + if plugin.routers is not None: + inject_app_router(plugin, main_router) return main_router +def resolve_plugin_order(plugins: list[PluginEntry]) -> list[PluginEntry]: + """ + 根据 depends_on 对插件排序 + + :param plugins: 插件配置列表 + :return: + """ + plugin_map = {plugin.name: plugin for plugin in plugins} + ordered_plugins: list[PluginEntry] = [] + visited: set[str] = set() + visiting: list[str] = [] + + def visit(plugin: PluginEntry) -> None: + if plugin.name in visited: + return + if plugin.name in visiting: + cycle_start = visiting.index(plugin.name) + cycle_path = [*visiting[cycle_start:], plugin.name] + raise PluginConfigError(f'插件存在循环依赖: {" -> ".join(cycle_path)}') + + if plugin.depends_on is not None: + visiting.append(plugin.name) + for dep_name in plugin.depends_on: + dep_plugin = plugin_map.get(dep_name) + if dep_plugin is None: + raise PluginConfigError(f'插件 {plugin.name} 依赖插件 {dep_name}，但插件 {dep_name} 不存在') + visit(dep_plugin) + visiting.pop() + + visited.add(plugin.name) + ordered_plugins.append(plugin) + + for plugin in plugins: + visit(plugin) + + return ordered_plugins + + def setup_plugins(app: FastAPI) -> None: """ 注册并执行插件 hooks @@ -374,32 +423,37 @@ def setup_plugins(app: FastAPI) -> None: :param app: FastAPI 应用实例 :return: """ - plugins = get_plugins() - enabled_plugins = get_enabled_plugins(plugins) + enabled_plugins = get_enabled_plugins() + extend_plugins, app_plugins = parse_plugin_config() + plugins: list[PluginEntry] = [plugin for plugin in extend_plugins + app_plugins if plugin.name in enabled_plugins] - for plugin in plugins: - if plugin not in enabled_plugins: - log.info(f'插件 {plugin} 未启用，已跳过 hooks 注册与执行') - continue + # 按依赖关系排序 + try: + ordered_plugins = resolve_plugin_order(plugins) + except PluginConfigError as e: + log.error(f'插件依赖解析失败: {e}') + raise - module_path = f'backend.plugin.{plugin}.hooks' + # 注册并执行 hooks + for plugin in ordered_plugins: + module_path = f'backend.plugin.{plugin.name}.hooks' try: module = import_module_cached(module_path) except ModuleNotFoundError as e: if e.name == module_path: - # 未定义 hooks.py continue - log.warning(f'插件 {plugin} hooks 模块加载失败: {e}') + log.warning(f'插件 {plugin.name} hooks 加载失败: {e}') continue except Exception as e: - log.warning(f'插件 {plugin} hooks 模块加载失败: {e}') + log.warning(f'插件 {plugin.name} hooks 加载失败: {e}') continue try: - register_plugin_lifespan_hook(plugin, module) - run_plugin_startup_hook(plugin, module, app) + register_plugin_lifespan_hook(plugin.name, module) + run_plugin_startup_hook(plugin.name, module, app) except Exception as e: - log.error(f'插件 {plugin} hooks 执行失败: {e}') + log.exception(f'插件 {plugin.name} hooks 执行失败: {e}') + raise PluginInjectError(f'插件 {plugin.name} hooks 执行失败：{e!s}') from e class PluginStatusChecker: diff --git a/backend/plugin/validator.py b/backend/plugin/validator.py index f921ca5e..5878a7fd 100644 --- a/backend/plugin/validator.py +++ b/backend/plugin/validator.py @@ -25,6 +25,7 @@ class PluginInfoSchema(BaseModel): author: str = Field(..., min_length=1, max_length=50, description='作者') tags: list[str] = Field(..., min_length=1, description='标签') database: list[str] = Field(..., min_length=1, description='数据库支持') + depends_on: list[str] = Field(default_factory=list, description='依赖的插件列表') @field_validator('version') @classmethod @@ -58,6 +59,15 @@ class PluginInfoSchema(BaseModel): ) return v + @field_validator('depends_on') + @classmethod + def validate_depends_on(cls, v: list[str]) -> list[str]: + """校验插件依赖配置""" + for dep in v: + if not dep or not isinstance(dep, str): + raise PluginConfigError(f'依赖的插件列表必须为非空字符串，当前值: {dep}') + return v + class AppPluginAppSchema(BaseModel): """应用级插件 app 配置模型""" @@ -183,6 +193,10 @@ def validate_plugin_config(plugin_name: str, config: dict[str, Any]) -> PluginLe error_msg = '; '.join(error_details) raise PluginConfigError(f'插件 {plugin_name} 配置校验失败: {error_msg}') from e + depends_on = config['plugin'].get('depends_on', []) + if plugin_name in depends_on: + raise PluginConfigError(f'插件 {plugin_name} 不能依赖自身') + plugin_dir = Path(PLUGIN_DIR) / plugin_name model_dir = plugin_dir / 'model' if model_dir.is_dir():