update plugin route inject

Author: wu-clan

backend/app/admin/service/data_rule_service.py

@@ -13,7 +13,7 @@
 from backend.core.conf import settings
 from backend.database.db import async_db_session
 from backend.database.redis import redis_client
-from backend.utils.import_parse import dynamic_import
+from backend.utils.import_parse import dynamic_import_data_model
 
 
 class DataRuleService:
@@ -43,7 +43,7 @@ async def get_columns(model: str) -> list[str]:
         if model not in settings.DATA_PERMISSION_MODELS:
             raise errors.NotFoundError(msg='数据模型不存在')
         try:
-            model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[model])
+            model_ins = dynamic_import_data_model(settings.DATA_PERMISSION_MODELS[model])
         except (ImportError, AttributeError):
             raise errors.ServerError(msg=f'数据模型 {model} 动态导入失败，请联系系统超级管理员')
         model_columns = [

backend/common/security/permission.py

@@ -9,7 +9,7 @@
 from backend.common.exception import errors
 from backend.common.exception.errors import ServerError
 from backend.core.conf import settings
-from backend.utils.import_parse import dynamic_import
+from backend.utils.import_parse import dynamic_import_data_model
 
 if TYPE_CHECKING:
     from backend.app.admin.schema.data_rule import GetDataRuleDetail
@@ -61,7 +61,7 @@ def filter_data_permission(request: Request) -> ColumnElement[bool]:
         if rule_model not in settings.DATA_PERMISSION_MODELS:
             raise errors.NotFoundError(msg='数据规则模型不存在')
         try:
-            model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[rule_model])
+            model_ins = dynamic_import_data_model(settings.DATA_PERMISSION_MODELS[rule_model])
         except (ImportError, AttributeError):
             raise errors.ServerError(msg=f'数据模型 {rule_model} 动态导入失败，请联系系统超级管理员')
         model_columns = [

backend/core/registrar.py

@@ -20,7 +20,7 @@
 from backend.middleware.jwt_auth_middleware import JwtAuthMiddleware
 from backend.middleware.opera_log_middleware import OperaLogMiddleware
 from backend.middleware.state_middleware import StateMiddleware
-from backend.plugin.tools import register_plugin_router
+from backend.plugin.tools import plugin_router_inject
 from backend.utils.demo_site import demo_site
 from backend.utils.health_check import ensure_unique_route_names, http_limit_callback
 from backend.utils.openapi import simplify_operation_ids
@@ -80,7 +80,6 @@ def register_app():
 
     # 路由
     register_router(app)
-    register_plugin_router(app)
 
     # 分页
     register_page(app)
@@ -160,7 +159,8 @@ def register_router(app: FastAPI):
     dependencies = [Depends(demo_site)] if settings.DEMO_MODE else None
 
     # API
-    app.include_router(router, dependencies=dependencies)
+    new_router = plugin_router_inject(router)
+    app.include_router(new_router, dependencies=dependencies)
 
     # Extra
     ensure_unique_route_names(app)

backend/plugin/notice/api/v1/__init__.py

@@ -0,0 +1,2 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-

backend/plugin/notice/api/v1/sys/__init__.py

@@ -0,0 +1,2 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-

backend/plugin/notice/api/notice.py renamed to backend/plugin/notice/api/v1/sys/notice.py

@@ -1,9 +1,8 @@
 # 属于哪个 app，如果为独立 app，应设置为 ''
 app = 'admin'
 
-# api 配置，仅针对于非独立 app
+# api 路由配置，仅对于非独立 app 可用
 [api]
-# api 注入模块，将生成对应路由
-module = 'v1.sys'
-# 路由 prefix 配置
-prefix = 'notices'
+# prefix 必须带前导 /
+prefix = '/notices'
+tags = '系统通知公告'

backend/plugin/notice/plugin.toml

@@ -1,14 +1,15 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+import importlib
 import inspect
 import os
+import warnings
 
 import rtoml
 
-from fastapi import FastAPI
+from fastapi import APIRouter
 
 from backend.core.path_conf import PLUGIN_DIR
-from backend.utils.import_parse import import_module_cached
 
 
 def get_plugins() -> list[str]:
@@ -31,59 +32,88 @@ def get_plugin_models() -> list:
     plugins = get_plugins()
     for plugin in plugins:
         module_path = f'backend.plugin.{plugin}.model'
-        module = import_module_cached(module_path)
+        module = importlib.import_module(module_path)
         for name, obj in inspect.getmembers(module):
             if inspect.isclass(obj):
                 classes.append(obj)
     return classes
 
 
-def register_plugin_router(app: FastAPI):
+def plugin_router_inject(router: APIRouter) -> APIRouter:
     """
     插件路由注入
 
-    :param app:
-    :return:
+    :param router: 源总路由器
+    :return: 注入插件路由后的总路由器
     """
     plugins = get_plugins()
     for plugin in plugins:
         toml_path = os.path.join(PLUGIN_DIR, plugin, 'plugin.toml')
         if not os.path.exists(toml_path):
-            raise FileNotFoundError('插件缺少 plugin.toml 配置文件')
+            raise FileNotFoundError(f'插件 {plugin} 缺少 plugin.toml 配置文件，请检查插件是否合法')
 
-        toml_data = rtoml.load(toml_path)
-        app_name = toml_data.get('app', '')
-        api_module = toml_data.get('api', {}).get('module', '')
-        prefix = toml_data.get('api', {}).get('prefix', '')
+        # 解析 plugin.toml
+        with open(toml_path, 'r', encoding='utf-8') as f:
+            data = rtoml.load(f)
+        app_name = data.get('app', '')
+        prefix = data.get('api', {}).get('prefix', '')
+        tags = data.get('api', {}).get('tags', [])
 
-        if app_name:
-            if '.' in api_module:
-                module_path = f'backend.app.{app_name}.api.{api_module}'
-            else:
-                module_path = f'backend.app.{app_name}.api'
-        else:
-            module_path = 'backend.app'
+        # 插件中 API 路由文件的路径
+        plugin_api_path = os.path.join(PLUGIN_DIR, plugin, 'api')
+        if not os.path.exists(plugin_api_path):
+            raise FileNotFoundError(f'插件 {plugin} 缺少 api 目录，请检查插件文件是否完整')
 
-        try:
-            module = import_module_cached(module_path)
-        except ImportError as e:
-            raise ImportError(f'导入模块 {module_path} 失败：{e}') from e
+        # 路由注入
+        if app_name:
+            # 非独立应用：将插件中的路由注入到源程序对应模块的路由中
+            for root, _, api_files in os.walk(plugin_api_path):
+                for file in api_files:
+                    if file.endswith('.py') and file != '__init__.py':
+                        api_files_path = os.path.join(root, file)
+
+                        # 获取插件路由模块
+                        path_to_module_str = os.path.relpath(api_files_path, PLUGIN_DIR).replace(os.sep, '.')[:-3]
+                        module_path = f'backend.plugin.{path_to_module_str}'
+                        try:
+                            module = importlib.import_module(module_path)
+                        except ImportError as e:
+                            raise ImportError(f'导入模块 {module_path} 失败：{e}') from e
+                        plugin_router = getattr(module, 'router', None)
+                        if not plugin_router:
+                            warnings.warn(
+                                f'目标模块 {module_path} 中没有有效的 router，请检查插件文件是否完整',
+                                FutureWarning,
+                            )
+                            continue
+
+                        # 获取源程序路由模块
+                        relative_path = os.path.relpath(root, plugin_api_path)
+                        target_module_path = f'backend.app.{app_name}.api.{relative_path.replace(os.sep, ".")}'
+                        try:
+                            target_module = importlib.import_module(target_module_path)
+                        except ImportError as e:
+                            raise ImportError(f'导入目标模块 {target_module_path} 失败：{e}') from e
+                        target_router = getattr(target_module, 'router', None)
+                        if not target_router or not isinstance(target_router, APIRouter):
+                            raise AttributeError(f'目标模块 {module_path} 中没有有效的 router，请检查插件文件是否完整')
+
+                        # 将插件路由注入到目标 router 中
+                        target_router.include_router(
+                            router=plugin_router,
+                            prefix=prefix,
+                            tags=tags if type(tags) is list else [tags],
+                        )
         else:
-            if app_name and '.' in api_module:
-                # 从 backend.app.xxx.api.vx.xxx.__init__.py 文件中获取路由
-                router = getattr(module, 'router', None)
-            else:
-                if 'api' in module_path:
-                    # 从 backend.app.xxx.api 下的 router.py 文件中获取路由
-                    router = getattr(module, api_module.split('.')[0], None)
-                else:
-                    # 从 backend.app 下的 router.py 文件中获取路由
-                    router = getattr(module, 'router', None)
-
-            if not router:
-                raise ImportError(f'模块 {module_path} 中不存在路由')
-
-            if app_name:
-                app.include_router(router, prefix=f'/{prefix}')
-            else:
-                app.include_router(router)
+            # 独立应用：将插件中的路由直接注入到 app 中
+            module_path = f'backend.plugin.{plugin}.api.router'
+            try:
+                target_module = importlib.import_module(module_path)
+            except ImportError as e:
+                raise ImportError(f'导入目标模块 {module_path} 失败：{e}') from e
+            target_router = getattr(target_module, 'router', None)
+            if not target_router or not isinstance(target_router, APIRouter):
+                raise AttributeError(f'目标模块 {module_path} 中没有有效的 router，请检查插件文件是否完整')
+            router.include_router(target_router)
+
+    return router

backend/plugin/tools.py

@@ -8,34 +8,34 @@
 
 def module_parse(module_path: str) -> tuple:
     """
-    Parse a module string into a Python module and class/function.
+    Parse a python module string into a python module and class/function.
 
     :param module_path:
     :return:
     """
-    module_name, class_or_func = module_path.rsplit('.', 1)
-    return module_name, class_or_func
+    module_path, class_or_func = module_path.rsplit('.', 1)
+    return module_path, class_or_func
 
 
 @lru_cache(maxsize=512)
-def import_module_cached(module_name: str) -> Any:
+def import_module_cached(module_path: str) -> Any:
     """
     缓存导入模块
 
-    :param module_name:
+    :param module_path:
     :return:
     """
-    return importlib.import_module(module_name)
+    return importlib.import_module(module_path)
 
 
-def dynamic_import(module_path: str) -> Any:
+def dynamic_import_data_model(module_path: str) -> Any:
     """
-    动态导入
+    动态导入数据模型
 
     :param module_path:
     :return:
     """
-    module_name, class_or_func = module_parse(module_path)
-    module = import_module_cached(module_name)
+    module_path, class_or_func = module_parse(module_path)
+    module = import_module_cached(module_path)
     ins = getattr(module, class_or_func)
     return ins