55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
|
"""repositories 层公共查询工具。
|
|||
|
|
|
|||
|
|
集中两类易错点，避免各 repository 各写一份、口径漂移：
|
|||
|
|
|
|||
|
|
- **模糊匹配的 LIKE 通配符转义**：直接 ``f"%{value}%"`` 拼接时，输入中的
|
|||
|
|
下划线会作为单字符通配执行（本系统用户名普遍含下划线，误配是常态而非
|
|||
|
|
理论），百分号则命中全表并触发全表扫描
|
|||
|
|
- **两类日志共有的筛选条件**：用户名/结果/时间范围在操作日志与登录日志
|
|||
|
|
之间逐字重复，抽为同一实现后「列表与计数」「操作与登录」不可能对不上
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from sqlalchemy import Select
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(frozen=True, slots=True)
|
|||
|
|
class LogFilters:
|
|||
|
|
"""两类日志（操作/登录）共有的筛选条件。"""
|
|||
|
|
|
|||
|
|
username: str | None = None
|
|||
|
|
success: int | None = None
|
|||
|
|
start_time: datetime | None = None
|
|||
|
|
end_time: datetime | None = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def fuzzy(stmt: Select, column: Any, value: str) -> Select:
|
|||
|
|
"""模糊匹配（含通配符自动转义）。
|
|||
|
|
|
|||
|
|
``contains(autoescape=True)`` 会转义输入中的 ``%``/``_`` 并附加 ``ESCAPE``
|
|||
|
|
子句，把用户输入当字面量匹配。
|
|||
|
|
"""
|
|||
|
|
return stmt.where(column.contains(value, autoescape=True))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def apply_log_filters(stmt: Select, model: Any, filters: LogFilters) -> Select:
|
|||
|
|
"""应用两类日志共有的筛选条件。
|
|||
|
|
|
|||
|
|
``model`` 传 ``OperationLog`` / ``LoginLog`` 的模型类（两者字段同名同型）。
|
|||
|
|
|
|||
|
|
列表与计数共用本函数，避免「列表按条件过滤而总数不按条件过滤」——
|
|||
|
|
漏改一侧时表现为翻到最后一页是空页或页码虚增，且没有任何报错。
|
|||
|
|
"""
|
|||
|
|
if filters.username:
|
|||
|
|
stmt = fuzzy(stmt, model.username, filters.username)
|
|||
|
|
if filters.success is not None:
|
|||
|
|
stmt = stmt.where(model.success == filters.success)
|
|||
|
|
if filters.start_time:
|
|||
|
|
stmt = stmt.where(model.create_time >= filters.start_time)
|
|||
|
|
if filters.end_time:
|
|||
|
|
stmt = stmt.where(model.create_time <= filters.end_time)
|
|||
|
|
return stmt
|