c83287819a
- 操作日志审计机制：audit 上下文（成功与失败都记、独立提交、操作人标量 固化、commit 后查询前置约定），用户/角色全部写操作接入；request_data 按键名的结构化脱敏 + UTF-8 字节边界截断 - 角色管理：CRUD 与启停；admin 禁删禁停、有绑定禁删（FOR UPDATE 锁行）、 编码创建后不可改、停用角色不可再分配 - 日志查询：操作/登录两端点，列表与详情两段式（详情含已脱敏请求参数）； 时间参数统一转本地 naive（修混用时区 500 与偏移丢弃） - 用户管理：补齐 reset_password 操作人、编辑空串归一 NULL、并发唯一键兜底 400 - 测试 301 例全绿（单元 223 / 集成 78） Co-Authored-By: Claude Code <noreply@anthropic.com>
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""日志查询编排。
|
||
|
||
操作日志与登录日志分别提供分页查询（两表字段与筛选项不同），操作日志另有
|
||
单条详情（列表为性能排除的 `request_data` 走详情按需加载）。
|
||
纯只读操作，不写审计 —— 读操作记日志只会淹没审计表。
|
||
"""
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.exceptions import NotFoundError
|
||
from app.core.response import PageResult
|
||
from app.repositories import login_log_repository, operation_log_repository
|
||
from app.repositories.query_utils import LogFilters
|
||
from app.schemas.log import (
|
||
LoginLogOut,
|
||
LoginLogQuery,
|
||
OperationLogDetailOut,
|
||
OperationLogOut,
|
||
OperationLogQuery,
|
||
)
|
||
|
||
|
||
def list_operation_logs(session: Session, query: OperationLogQuery) -> PageResult[OperationLogOut]:
|
||
"""分页查询操作日志（列表不含 request_data，详情走 `get_operation_log`）。"""
|
||
filters = _filters_from(query)
|
||
logs = operation_log_repository.list_logs(
|
||
session,
|
||
offset=query.offset,
|
||
limit=query.page_size,
|
||
filters=filters,
|
||
module=query.module,
|
||
action=query.action,
|
||
)
|
||
total = operation_log_repository.count_logs(
|
||
session, filters=filters, module=query.module, action=query.action
|
||
)
|
||
return PageResult[OperationLogOut](
|
||
records=[OperationLogOut.model_validate(log) for log in logs],
|
||
total=total,
|
||
page_num=query.page_num,
|
||
page_size=query.page_size,
|
||
)
|
||
|
||
|
||
def get_operation_log(session: Session, log_id: int) -> OperationLogDetailOut:
|
||
"""查询单条操作日志详情（含已脱敏的 request_data）。
|
||
|
||
:raises NotFoundError: 日志不存在
|
||
"""
|
||
log = operation_log_repository.get_by_id(session, log_id)
|
||
if log is None:
|
||
raise NotFoundError("日志不存在")
|
||
return OperationLogDetailOut.model_validate(log)
|
||
|
||
|
||
def list_login_logs(session: Session, query: LoginLogQuery) -> PageResult[LoginLogOut]:
|
||
"""分页查询登录日志。"""
|
||
filters = _filters_from(query)
|
||
logs = login_log_repository.list_logs(
|
||
session,
|
||
offset=query.offset,
|
||
limit=query.page_size,
|
||
filters=filters,
|
||
ip=query.ip,
|
||
)
|
||
total = login_log_repository.count_logs(session, filters=filters, ip=query.ip)
|
||
return PageResult[LoginLogOut](
|
||
records=[LoginLogOut.model_validate(log) for log in logs],
|
||
total=total,
|
||
page_num=query.page_num,
|
||
page_size=query.page_size,
|
||
)
|
||
|
||
|
||
def _filters_from(query: OperationLogQuery | LoginLogQuery) -> LogFilters:
|
||
"""两类 Query 的公共筛选字段 → LogFilters（口径与 repository 共用一个实现）。"""
|
||
return LogFilters(
|
||
username=query.username,
|
||
success=query.success,
|
||
start_time=query.start_time,
|
||
end_time=query.end_time,
|
||
)
|