Optimize the dynamic import of data models (#546)

Author: wu-clan

backend/app/admin/service/data_rule_service.py

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

backend/core/conf.py

@@ -169,7 +169,7 @@ class Settings(BaseSettings):
     DATA_PERMISSION_MODELS: dict[
         str, str
     ] = {  # 允许进行数据过滤的 SQLA 模型，它必须以模块字符串的方式定义（它应该只用于前台数据，这里只是为了演示）
-        'Api': 'backend.app.admin.model.Api',
+        'Api': 'backend.plugin.casbin.model.Api',
     }
     DATA_PERMISSION_COLUMN_EXCLUDE: list[str] = [  # 排除允许进行数据过滤的 SQLA 模型列
         'id',

backend/utils/import_parse.py

@@ -5,16 +5,8 @@
 from functools import lru_cache
 from typing import Any
 
-
-def module_parse(module_path: str) -> tuple:
-    """
-    Parse a python module string into a python module and class/function.
-
-    :param module_path:
-    :return:
-    """
-    module_path, class_or_func = module_path.rsplit('.', 1)
-    return module_path, class_or_func
+from backend.common.exception import errors
+from backend.common.log import log
 
 
 @lru_cache(maxsize=512)
@@ -35,7 +27,12 @@ def dynamic_import_data_model(module_path: str) -> Any:
     :param module_path:
     :return:
     """
-    module_path, class_or_func = module_parse(module_path)
-    module = import_module_cached(module_path)
-    ins = getattr(module, class_or_func)
+    module_path, class_or_func = module_path.rsplit('.', 1)
+
+    try:
+        module = import_module_cached(module_path)
+        ins = getattr(module, class_or_func)
+    except (ImportError, AttributeError) as e:
+        log.error(e)
+        raise errors.ServerError(msg='数据模型列动态解析失败，请联系系统超级管理员')
     return ins