Add plugin model alembic support

Author: wu-clan

backend/alembic/env.py

@@ -16,11 +16,20 @@
 from backend.common.model import MappedBase
 from backend.core import path_conf
 from backend.database.db import SQLALCHEMY_DATABASE_URL
+from backend.plugin.plugin import get_plugin_models
 
 # import your new model here
 from backend.app.admin.model import *  # noqa: F401
 from backend.app.generator.model import *  # noqa: F401
 
+# import plugin model
+for cls in get_plugin_models():
+    class_name = cls.__name__
+    if class_name in globals():
+        print(f'\nWarning: Class "{class_name}" already exists in global namespace.')
+    else:
+        globals()[class_name] = cls
+
 if not os.path.exists(path_conf.ALEMBIC_VERSION_DIR):
     os.makedirs(path_conf.ALEMBIC_VERSION_DIR)
 

backend/app/admin/service/data_rule_service.py

@@ -42,7 +42,10 @@ async def get_models() -> list[str]:
     async def get_columns(model: str) -> list[str]:
         if model not in settings.DATA_PERMISSION_MODELS:
             raise errors.NotFoundError(msg='数据模型不存在')
-        model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[model])
+        try:
+            model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[model])
+        except (ImportError, AttributeError):
+            raise errors.ServerError(msg=f'数据模型 {model} 动态导入失败，请联系系统超级管理员')
         model_columns = [
             key for key in model_ins.__table__.columns.keys() if key not in settings.DATA_PERMISSION_COLUMN_EXCLUDE
         ]

backend/common/security/permission.py

@@ -60,7 +60,10 @@ def filter_data_permission(request: Request) -> ColumnElement[bool]:
         rule_model = rule.model
         if rule_model not in settings.DATA_PERMISSION_MODELS:
             raise errors.NotFoundError(msg='数据规则模型不存在')
-        model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[rule_model])
+        try:
+            model_ins = dynamic_import(settings.DATA_PERMISSION_MODELS[rule_model])
+        except (ImportError, AttributeError):
+            raise errors.ServerError(msg=f'数据模型 {rule_model} 动态导入失败，请联系系统超级管理员')
         model_columns = [
             key for key in model_ins.__table__.columns.keys() if key not in settings.DATA_PERMISSION_COLUMN_EXCLUDE
         ]

backend/core/path_conf.py

@@ -22,3 +22,6 @@
 
 # jinja2 模版文件路径
 JINJA2_TEMPLATE_DIR = os.path.join(BasePath, 'templates')
+
+# 插件目录
+PLUGIN_DIR = os.path.join(BasePath, 'plugin')

backend/plugin/__init__.py

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

backend/plugin/conf.toml

@@ -1,5 +1,3 @@
-# 插件名称
-name = 'notice'
 # 是否对此插件启用 alembic
 alembic = true
 

backend/plugin/notice/plugin.toml

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import inspect
+import os
+
+from backend.core.path_conf import PLUGIN_DIR
+from backend.utils.import_parse import import_module_cached
+
+
+def get_plugins() -> list[str]:
+    """获取插件"""
+    plugin_packages = []
+
+    for item in os.listdir(PLUGIN_DIR):
+        item_path = os.path.join(PLUGIN_DIR, item)
+
+        if os.path.isdir(item_path):
+            if '__init__.py' in os.listdir(item_path):
+                plugin_packages.append(item)
+
+    return plugin_packages
+
+
+def get_plugin_models():
+    """获取插件所有模型类"""
+    classes = []
+    plugins = get_plugins()
+    for plugin in plugins:
+        module_path = f'backend.plugin.{plugin}.model'
+        module = import_module_cached(module_path)
+        for name, obj in inspect.getmembers(module):
+            if inspect.isclass(obj):
+                classes.append(obj)
+    return classes
+
+
+if __name__ == '__main__':
+    print(get_plugins())
+    print(get_plugin_models())

backend/plugin/plugin.py

@@ -5,10 +5,8 @@
 from functools import lru_cache
 from typing import Any
 
-from backend.common.exception import errors
 
-
-def parse_module_str(module_path: str) -> tuple:
+def module_parse(module_path: str) -> tuple:
     """
     Parse a module string into a Python module and class/function.
 
@@ -37,11 +35,7 @@ def dynamic_import(module_path: str) -> Any:
     :param module_path:
     :return:
     """
-    module_name, obj_name = parse_module_str(module_path)
-
-    try:
-        module = import_module_cached(module_name)
-        class_or_func = getattr(module, obj_name)
-        return class_or_func
-    except (ImportError, AttributeError):
-        raise errors.ServerError(msg=f'数据模型 {module_name} 动态导入失败，请联系系统超级管理员')
+    module_name, class_or_func = module_parse(module_path)
+    module = import_module_cached(module_name)
+    ins = getattr(module, class_or_func)
+    return ins