Skip to content

操作日志

操作日志通过 @operation_log 装饰器标记需要记录日志的视图方法,自动记录请求参数、响应结果、耗时、操作类型等信息。源码位于 application/operation_log/decorators.py

工作原理

1. 视图方法通过 @operation_log(title=..., log_type=LogType.XXX) 标记
2. 装饰器在方法执行前记录开始时间、提取请求参数
3. 执行原方法,获取响应
4. 检查响应是否成功(解析 JSON 中的 code 字段)
5. 记录操作日志到数据库(含参数、结果、耗时、状态)
6. 异常时记录错误信息并重新抛出

装饰器使用

python
# application/level/views.py
from application.operation_log.decorators import operation_log
from application.operation_log.constants import LogType

class LevelAddView(PermissionRequired, View):
    permission_required = ("sys:level:add",)

    @operation_log(
        title="职级管理-添加记录",
        log_type=LogType.ADD
    )
    def post(self, request):
        result = services.add_level(request)
        return result

装饰器参数

参数类型说明
titlestr操作标题,如"职级管理-添加职级"
log_typeint操作类型,参考 LogType 常量
sourceint操作来源,0-后台系统 1-手机站 2-客户端 3-其他

LogType 常量

python
# application/operation_log/constants.py
class LogType:
    OTHER = 0        # 其它
    ADD = 1           # 新增
    UPDATE = 2        # 修改
    DELETE = 3        # 删除
    BATCH_DELETE = 4  # 批量删除
    STATUS = 5        # 状态设置
    RESET_PWD = 6     # 重置密码
    UPDATE_PWD = 7    # 修改密码
    IMPORT = 8        # 导入
    EXPORT = 9        # 导出
    QUERY = 10        # 查询

核心实现

python
# application/operation_log/decorators.py
def operation_log(title: str, log_type: int = LogType.OTHER, source: int = 0):
    def decorator(func):
        @wraps(func)
        def wrapper(self, request, *args, **kwargs):
            # 记录开始时间
            start_time = time.time()

            # 保存请求参数用于日志记录
            request.log_params = _extract_params(request)

            try:
                # 执行原方法
                response = func(self, request, *args, **kwargs)

                # 计算耗时(毫秒)
                consume_time = int((time.time() - start_time) * 1000)

                # 检查响应是否成功
                is_success, error_msg = _check_response_success(response)
                status = 0 if is_success else 1

                # 记录操作日志
                record_operation_log(
                    request=request,
                    title=title,
                    log_type=log_type,
                    source=source,
                    status=status,
                    error_msg=error_msg,
                    consume_time=consume_time,
                    result=response
                )

                return response

            except Exception as e:
                consume_time = int((time.time() - start_time) * 1000)
                record_operation_log(
                    request=request,
                    title=title,
                    log_type=log_type,
                    source=source,
                    status=1,
                    error_msg=str(e),
                    consume_time=consume_time,
                    result=None
                )
                raise

        return wrapper
    return decorator

参数提取

装饰器自动提取请求参数,支持 GET/POST/PUT/DELETE/PATCH:

python
def _extract_params(request):
    """提取请求参数,自动过滤敏感信息"""
    params = {}

    if request.method == 'GET':
        params = request.GET.dict()
    elif request.method in ['POST', 'PUT', 'DELETE', 'PATCH']:
        try:
            params = json.loads(request.body.decode('utf-8'))
        except:
            params = request.POST.dict()

    # 过滤敏感信息
    sensitive_keys = ['password', 'pwd', 'token', 'secret', 'access_token']
    for key in sensitive_keys:
        if key in params:
            params[key] = '******'

    return params

响应检查

装饰器自动检查响应是否成功:

python
def _check_response_success(response):
    """检查响应是否成功"""
    is_success = True
    error_msg = None

    try:
        # 检查 HTTP 状态码
        if hasattr(response, 'status_code') and response.status_code >= 400:
            return False, f"HTTP {response.status_code}"

        # 解析 JSON 响应内容
        if hasattr(response, 'content') and response.content:
            content = response.content.decode('utf-8')
            data = json.loads(content)
            if isinstance(data, dict):
                code = data.get('code')
                if code is not None and code != 0:
                    is_success = False
                    error_msg = data.get('msg', '操作失败')
    except:
        pass

    return is_success, error_msg

日志字段

字段说明
title操作标题
log_type操作类型
url请求路径
method请求方法
param请求参数(截取前2000字符)
result响应结果(截取前2000字符)
consume_time耗时(毫秒)
status状态:0-成功 1-失败
error_msg错误信息
create_user操作人

查询接口

GET /operation/log/page?pageNo=1&pageSize=10
GET /operation/log/detail/<int:log_id>

总结

操作日志模块具备以下特点:

1. 装饰器模式:@operation_log 标记需要记录日志的视图方法
2. 自动记录:参数、结果、耗时、操作人等完整信息
3. 类型常量:LogType 定义标准操作类型
4. 敏感过滤:自动过滤密码、token 等敏感信息
5. 异常捕获:异常时记录错误信息并重新抛出
6. 响应检查:自动解析 JSON 响应判断操作是否成功

小蚂蚁云团队 · 提供技术支持