bwadmin/pear-admin-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

基本完成文档

Browse Source
This commit is contained in:
wojiaoyishang
2025-01-30 10:31:50 +08:00
parent 41bda82c60
commit ac1f82e81d
17 changed files with 1192 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
applications/common/helper.py
+91 -63
View File
@@ -1,10 +1,10 @@
from sqlalchemy import and_
from sqlalchemy import and_, func
from applications.extensions import db
class ModelFilter:
"""
ORM 多条件查询构造器，支持多种查询条件组合。
ORM 多条件查询构造器，支持多种查询条件组合，自动转义特殊字符防止SQL注入。
示例：
mf = ModelFilter()
@@ -13,14 +13,14 @@ class ModelFilter:
query = User.query.filter(mf.get_filter(User))
"""
filter_field = {} # 存储字段过滤条件
filter_list = [] # 存储最终的过滤条件列表
filter_list = [] # 存储最终的过滤条件列表
# 查询类型常量
type_exact = "exact" # 精确匹配
type_neq = "neq" # 不等于
type_exact = "exact" # 精确匹配
type_neq = "neq" # 不等于
type_greater = "greater" # 大于
type_less = "less" # 小于
type_vague = "vague" # 模糊匹配
type_less = "less" # 小于
type_vague = "vague" # 模糊匹配
type_contains = "contains" # 包含
type_between = "between" # 范围查询
@@ -29,117 +29,145 @@ class ModelFilter:
self.filter_field = {}
self.filter_list = []
@staticmethod
def escape_like(value: str, escape_char: str = '\\') -> str:
"""
转义LIKE查询中的特殊字符（%, _ 和转义字符本身）
:param value: 需要转义的原始字符串
:param escape_char: 转义字符（默认反斜杠）
:return: 转义后的安全字符串
"""
return (
value.replace(escape_char, escape_char * 2)
.replace('%', escape_char + '%')
.replace('_', escape_char + '_')
)
def exact(self, field_name, value):
"""
添加精确匹配条件。
添加精确匹配条件（自动处理字符串类型参数）
:param field_name: 模型字段名称。
:param value: 匹配的值。
:param field_name: 模型字段名称
:param value: 匹配的值（自动过滤空字符串）
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_exact}
if value is not None and value != '':
# 字符串类型自动调用escape_like（防止特殊字符影响精确匹配）
processed_value = self.escape_like(str(value)) if isinstance(value, str) else value
self.filter_field[field_name] = {"data": processed_value, "type": self.type_exact}
def neq(self, field_name, value):
"""
添加不等于条件。
添加不等于条件（自动处理字符串类型参数）
:param field_name: 模型字段名称。
:param value: 不匹配的值。
:param field_name: 模型字段名称
:param value: 不匹配的值（自动过滤空字符串）
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_neq}
if value is not None and value != '':
# 字符串类型自动调用escape_like
processed_value = self.escape_like(str(value)) if isinstance(value, str) else value
self.filter_field[field_name] = {"data": processed_value, "type": self.type_neq}
def greater(self, field_name, value):
"""
添加大于条件。
添加大于条件（数值/日期比较）
:param field_name: 模型字段名称。
:param value: 大于的值。
:param field_name: 模型字段名称
:param value: 比较的数值/日期
"""
if value and value != '':
if value is not None and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_greater}
def less(self, field_name, value):
"""
添加小于条件。
添加小于条件（数值/日期比较）
:param field_name: 模型字段名称。
:param value: 小于的值。
:param field_name: 模型字段名称
:param value: 比较的数值/日期
"""
if value and value != '':
if value is not None and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_less}
def vague(self, field_name, value: str):
"""
添加模糊匹配条件（左右模糊）。
添加安全模糊匹配（自动转义特殊字符，左右加%）
:param field_name: 模型字段名称。
:param value: 模糊匹配的值。
:param field_name: 模型字段名称
:param value: 需要模糊匹配的字符串（自动过滤空值）
"""
if value and value != '':
self.filter_field[field_name] = {"data": ('%' + value + '%'), "type": self.type_vague}
escaped_value = self.escape_like(value)
self.filter_field[field_name] = {"data": f'%{escaped_value}%', "type": self.type_vague}
def left_vague(self, field_name, value: str):
"""
添加左模糊匹配条件。
添加安全左模糊匹配（自动转义特殊字符，左侧加%）
:param field_name: 模型字段名称。
:param value: 左模糊匹配的值。
:param field_name: 模型字段名称
:param value: 需要左模糊匹配的字符串
"""
if value and value != '':
self.filter_field[field_name] = {"data": ('%' + value), "type": self.type_vague}
escaped_value = self.escape_like(value)
self.filter_field[field_name] = {"data": f'%{escaped_value}', "type": self.type_vague}
def right_vague(self, field_name, value: str):
"""
添加右模糊匹配条件。
添加安全右模糊匹配（自动转义特殊字符，右侧加%）
:param field_name: 模型字段名称。
:param value: 右模糊匹配的值。
:param field_name: 模型字段名称
:param value: 需要右模糊匹配的字符串
"""
if value and value != '':
self.filter_field[field_name] = {"data": (value + '%'), "type": self.type_vague}
escaped_value = self.escape_like(value)
self.filter_field[field_name] = {"data": f'{escaped_value}%', "type": self.type_vague}
def contains(self, field_name, value: str):
"""
添加包含条件。
添加安全包含条件（自动转义特殊字符，等效于vague）
:param field_name: 模型字段名称。
:param value: 包含的值。
:param field_name: 模型字段名称
:param value: 需要包含的字符串
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_contains}
escaped_value = self.escape_like(value)
self.filter_field[field_name] = {"data": f'%{escaped_value}%', "type": self.type_contains}
def between(self, field_name, value1, value2):
"""
添加范围查询条件。
添加范围查询条件（自动过滤无效值）
:param field_name: 模型字段名称。
:param value1: 范围起始值。
:param value2: 范围结束值。
:param field_name: 模型字段名称
:param value1: 范围起始值
:param value2: 范围结束值
"""
if value1 and value2 and value1 != '' and value2 != '':
if all([v is not None and v != '' for v in [value1, value2]]):
self.filter_field[field_name] = {"data": [value1, value2], "type": self.type_between}
def get_filter(self, model: db.Model):
"""
获取最终的 SQLAlchemy 过滤条件。
生成安全的SQLAlchemy过滤条件
:param model: SQLAlchemy 模型类。
:return: 返回组合后的过滤条件。
:param model: SQLAlchemy 模型类
:return: 组合后的过滤条件（使用and_连接）
"""
for k, v in self.filter_field.items():
if v.get("type") == self.type_vague:
self.filter_list.append(getattr(model, k).like(v.get("data")))
if v.get("type") == self.type_contains:
self.filter_list.append(getattr(model, k).contains(v.get("data")))
if v.get("type") == self.type_exact:
self.filter_list.append(getattr(model, k) == v.get("data"))
if v.get("type") == self.type_neq:
self.filter_list.append(getattr(model, k) != v.get("data"))
if v.get("type") == self.type_greater:
self.filter_list.append(getattr(model, k) > v.get("data"))
if v.get("type") == self.type_less:
self.filter_list.append(getattr(model, k) < v.get("data"))
if v.get("type") == self.type_between:
self.filter_list.append(getattr(model, k).between(v.get("data")[0], v.get("data")[1]))
field = getattr(model, k)
data = v.get("data")
query_type = v.get("type")
if query_type == self.type_vague:
self.filter_list.append(field.like(data, escape='\\'))
elif query_type == self.type_contains:
self.filter_list.append(field.like(data, escape='\\'))
elif query_type == self.type_exact:
self.filter_list.append(field == data)
elif query_type == self.type_neq:
self.filter_list.append(field != data)
elif query_type == self.type_greater:
self.filter_list.append(field > data)
elif query_type == self.type_less:
self.filter_list.append(field < data)
elif query_type == self.type_between:
self.filter_list.append(field.between(data[0], data[1]))
return and_(*self.filter_list)