update filter_data_permission

Author: wu-clan

backend/app/admin/schema/data_rule.py

@@ -36,7 +36,3 @@ class GetDataRuleDetail(DataRuleSchemaBase):
     id: int = Field(description='规则 ID')
     created_time: datetime = Field(description='创建时间')
     updated_time: datetime | None = Field(None, description='更新时间')
-
-    def __hash__(self) -> int:
-        """计算哈希值"""
-        return hash(self.name)

backend/common/security/permission.py

@@ -4,15 +4,19 @@
 
 from fastapi import Request
 from sqlalchemy import ColumnElement, and_, or_
+from sqlalchemy.ext.asyncio import AsyncSession
 
+from backend.app.admin.crud.crud_data_scope import data_scope_dao
 from backend.common.enums import RoleDataRuleExpressionType, RoleDataRuleOperatorType
 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_data_model
 
 if TYPE_CHECKING:
-    from backend.app.admin.schema.data_rule import GetDataRuleDetail
+    from backend.app.admin.model import DataRule
+
+    pass
 
 
 class RequestPermission:
@@ -47,33 +51,38 @@ async def __call__(self, request: Request) -> None:
             request.state.permission = self.value
 
 
-def filter_data_permission(request: Request) -> ColumnElement[bool]:
+async def filter_data_permission(db: AsyncSession, request: Request) -> ColumnElement[bool]:
     """
     过滤数据权限，控制用户可见数据范围
 
     使用场景：
-        - 用户登录前台后，控制其能看到哪些数据
-        - 根据用户角色和规则过滤数据访问权限
+        - 控制用户能看到哪些数据
 
+    :param db: 数据库会话
     :param request: FastAPI 请求对象
     :return:
     """
-    # 获取用户角色和规则
-    data_rules = []
+    # 获取用户角色和数据范围
+    data_scopes = []
     for role in request.user.roles:
-        data_rules.extend(role.rules)
-    user_data_rules: list[GetDataRuleDetail] = list(dict.fromkeys(data_rules))
+        data_scopes.extend(role.scopes)
 
     # 超级管理员和无规则用户不做过滤
-    if request.user.is_superuser or not user_data_rules:
+    if request.user.is_superuser or not data_scopes:
         return or_(1 == 1)
 
+    # 获取数据范围规则
+    data_rule_list: list[DataRule] = []
+    for data_scope in data_scopes:
+        data_scope_with_relation = await data_scope_dao.get_with_relation(db, data_scope.id)
+        data_rule_list.extend(data_scope_with_relation.rules)
+
     where_and_list = []
     where_or_list = []
 
-    for rule in user_data_rules:
+    for data_rule in list(dict.fromkeys(data_rule_list)):
         # 验证规则模型
-        rule_model = rule.model
+        rule_model = data_rule.model
         if rule_model not in settings.DATA_PERMISSION_MODELS:
             raise errors.NotFoundError(msg='数据规则模型不存在')
         model_ins = dynamic_import_data_model(settings.DATA_PERMISSION_MODELS[rule_model])
@@ -82,38 +91,38 @@ def filter_data_permission(request: Request) -> ColumnElement[bool]:
         model_columns = [
             key for key in model_ins.__table__.columns.keys() if key not in settings.DATA_PERMISSION_COLUMN_EXCLUDE
         ]
-        column = rule.column
+        column = data_rule.column
         if column not in model_columns:
             raise errors.NotFoundError(msg='数据规则模型列不存在')
 
         # 构建过滤条件
         column_obj = getattr(model_ins, column)
-        rule_expression = rule.expression
+        rule_expression = data_rule.expression
         condition = None
         if rule_expression == RoleDataRuleExpressionType.eq:
-            condition = column_obj == rule.value
+            condition = column_obj == data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.ne:
-            condition = column_obj != rule.value
+            condition = column_obj != data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.gt:
-            condition = column_obj > rule.value
+            condition = column_obj > data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.ge:
-            condition = column_obj >= rule.value
+            condition = column_obj >= data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.lt:
-            condition = column_obj < rule.value
+            condition = column_obj < data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.le:
-            condition = column_obj <= rule.value
+            condition = column_obj <= data_rule.value
         elif rule_expression == RoleDataRuleExpressionType.in_:
-            values = rule.value.split(',') if isinstance(rule.value, str) else rule.value
+            values = data_rule.value.split(',') if isinstance(data_rule.value, str) else data_rule.value
             condition = column_obj.in_(values)
-        elif rule.expression == RoleDataRuleExpressionType.not_in:
-            values = rule.value.split(',') if isinstance(rule.value, str) else rule.value
+        elif data_rule.expression == RoleDataRuleExpressionType.not_in:
+            values = data_rule.value.split(',') if isinstance(data_rule.value, str) else data_rule.value
             condition = ~column_obj.in_(values)
 
         # 根据运算符添加到对应列表
         if condition is not None:
-            if rule.operator == RoleDataRuleOperatorType.AND:
+            if data_rule.operator == RoleDataRuleOperatorType.AND:
                 where_and_list.append(condition)
-            elif rule.operator == RoleDataRuleOperatorType.OR:
+            elif data_rule.operator == RoleDataRuleOperatorType.OR:
                 where_or_list.append(condition)
 
     # 组合所有条件

backend/plugin/casbin/crud/crud_api.py

@@ -7,7 +7,6 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy_crud_plus import CRUDPlus
 
-from backend.common.security.permission import filter_data_permission
 from backend.plugin.casbin.model import Api
 from backend.plugin.casbin.schema.api import CreateApiParam, UpdateApiParam
 
@@ -43,7 +42,7 @@ async def get_list(self, request: Request, name: str = None, method: str = None,
         if path is not None:
             filters.update(path__like=f'%{path}%')
         stmt = await self.select_order('created_time', 'desc', **filters)
-        return stmt.where(filter_data_permission(request))
+        return stmt
 
     async def get_all(self, db: AsyncSession) -> Sequence[Api]:
         """