1- from functools import partial
2- from typing import Annotated
1+ from typing import Any
32
4- from fastapi import Depends , Request
3+ from fastapi import Request
54from sqlalchemy import ColumnElement , and_ , or_
65
76from backend .common .context import ctx
@@ -97,25 +96,34 @@ def filter_data_permission(request: Request, *models: object) -> ColumnElement[b
9796
9897 # 构建过滤条件
9998 column_obj = getattr (target_model , rule_column )
99+ column_type = target_model .__table__ .columns [rule_column ].type .python_type
100+
101+ def cast_value (value : Any ) -> Any :
102+ """类型转换"""
103+ try :
104+ return column_type (value ) if column_type is not str else value
105+ except (ValueError , TypeError ):
106+ return value
107+
100108 condition = None
101109 match data_rule .expression :
102110 case RoleDataRuleExpressionType .eq :
103- condition = column_obj == data_rule .value
111+ condition = column_obj == cast_value ( data_rule .value )
104112 case RoleDataRuleExpressionType .ne :
105- condition = column_obj != data_rule .value
113+ condition = column_obj != cast_value ( data_rule .value )
106114 case RoleDataRuleExpressionType .gt :
107- condition = column_obj > data_rule .value
115+ condition = column_obj > cast_value ( data_rule .value )
108116 case RoleDataRuleExpressionType .ge :
109- condition = column_obj >= data_rule .value
117+ condition = column_obj >= cast_value ( data_rule .value )
110118 case RoleDataRuleExpressionType .lt :
111- condition = column_obj < data_rule .value
119+ condition = column_obj < cast_value ( data_rule .value )
112120 case RoleDataRuleExpressionType .le :
113- condition = column_obj <= data_rule .value
121+ condition = column_obj <= cast_value ( data_rule .value )
114122 case RoleDataRuleExpressionType .in_ :
115- values = data_rule . value . split ( ',' ) if isinstance ( data_rule . value , str ) else data_rule .value
123+ values = [ cast_value ( v . strip ()) for v in data_rule .value . split ( ',' )]
116124 condition = column_obj .in_ (values )
117125 case RoleDataRuleExpressionType .not_in :
118- values = data_rule . value . split ( ',' ) if isinstance ( data_rule . value , str ) else data_rule .value
126+ values = [ cast_value ( v . strip ()) for v in data_rule .value . split ( ',' )]
119127 condition = column_obj .not_in (values )
120128
121129 # 根据运算符添加到对应列表
@@ -136,11 +144,22 @@ def filter_data_permission(request: Request, *models: object) -> ColumnElement[b
136144 return or_ (* where_list ) if where_list else or_ (1 == 1 )
137145
138146
139- def DataPermissionFilter (* models : object ) -> type [ColumnElement [bool ]]: # noqa: N802
140- """
141- 指定模型的数据权限过滤器
147+ # 此函数是为了简化调用方式,但目前无法正常工作: https://github.com/fastapi/fastapi/discussions/14438
148+ # def DataPermissionFilter(*models: object) -> type[ColumnElement[bool]]:
149+ # """
150+ # 指定模型的数据权限过滤器
151+ #
152+ # :param models: 模型类(可选,支持多个)
153+ # :return:
154+ # """
155+ # return Annotated[ColumnElement[bool], Depends(partial(filter_data_permission, *models))]
142156
143- :param models: 模型类(可选,支持多个)
144- :return:
145- """
146- return Annotated [ColumnElement [bool ], Depends (partial (filter_data_permission , * models ))]
157+
158+ class DataPermissionFilter :
159+ """指定模型的数据权限过滤器"""
160+
161+ def __init__ (self , * models : object ) -> None :
162+ self .models = models
163+
164+ async def __call__ (self , request : Request ) -> ColumnElement [bool ]:
165+ return filter_data_permission (request , * self .models )
0 commit comments