如何通过Python装饰器实现审计日志，将用户行为记录到数据库？
- 内容介绍
- 文章标签
- 相关推荐
本文共计1166个文字，预计阅读时间需要5分钟。
在业务函数上直接增加一层装饰器，是轻量级、侵入性最小的日志记录方案。它不改变原有的逻辑，仅负责将谁、什么时间、调用了什么、传了哪些参数、结果如何等信息记录到数据库。
常见错误是把所有参数原样 json.dumps 存进去——遇到 datetime、Decimal 或自定义对象会直接报 TypeError: Object of type ... is not JSON serializable。
- 只序列化可安全落库的字段：用户ID（
current_user.id）、函数名（func.__name__）、开始时间（datetime.utcnow()）、参数快照（用repr()或白名单过滤后转字典） - 避免记录敏感参数：比如密码、token、银行卡号，装饰器里加个
exclude_keys=('password', 'token', 'card_number')参数 - 别在装饰器里做同步 DB 写入——高并发下会拖慢主流程；用
threading.Thread或异步队列（如queue.Queue配后台线程）异步落库
log_operation 必须能拿到当前用户上下文
Web 场景下，用户信息通常存在请求上下文（如 Flask 的 g.user、Django 的 request.user），但装饰器本身没 request 对象。硬写 g.user 会导致非 Web 环境（如 Celery 任务、命令行脚本）直接崩。
典型报错：RuntimeError: Working outside of application context（Flask）或 AttributeError: 'NoneType' object has no attribute 'user'（Django）。
立即学习“Python免费学习笔记（深入）”；
- 装饰器不主动取用户，改为由被装饰函数显式传入：
@log_operation(user_id=user.id) - 或统一约定一个上下文获取函数（如
get_current_user_id()），在不同环境里各自实现：Web 环境从 request 取，CLI 环境返回0或'cli'，测试环境 mock 返回固定值 - 永远不要假设
current_user是全局可用的变量——它只是框架在特定生命周期内注入的临时对象
审计日志字段设计要区分「操作行为」和「数据变更」
只记「调用了 update_profile」不够，得知道「把邮箱从 a@old.com 改成了 b@new.com」。但直接存整个 model 实例 diff 成本高、易出错。
常见翻车点：用 model_to_dict(instance) 记录前后状态，结果发现外键字段是对象而非 ID，序列化失败；或者忘记排除 updated_at 这类自动更新字段，导致每次 diff 都不一样。
- 审计表至少包含：
user_id、operation（如'update_user_email'）、target_id（被操作对象 ID）、before_data（JSON 字段，仅含关键变更前字段）、after_data（同理）、created_at - diff 逻辑收在单独函数里，比如
diff_fields(instance, fields=['email', 'phone'])，只比对业务关心的字段 - 数据库字段类型选
JSONB（PostgreSQL）或TEXT（MySQL），别用JSON类型——有些旧版 MySQL 不支持，且查询能力弱
异步写日志时注意事务与连接泄漏
用 threading.Thread(target=save_log, args=(...)) 启动新线程写 DB，容易在 SQLAlchemy 场景下触发 DetachedInstanceError 或连接池耗尽——因为主线程的 session 已关闭，子线程还在用。
更隐蔽的问题：日志写入失败 silent fail，线上完全看不到记录，审计就形同虚设。
- 子线程内新建独立 session（
sessionmaker()()），不用主线程的 session - 必须加 try/except 包裹 DB 操作，并记录错误到
logging.error；失败日志至少要包含operation、user_id、error三要素 - 别依赖线程池或全局连接池管理——简单场景用单例队列 + 定期 flush 更稳；复杂系统再上 Celery 或 Kafka
真正难的不是怎么记，是怎么保证每条关键操作都逃不过日志——漏一条，审计链就断了。最常被绕过的其实是 ORM 的 bulk 操作、raw SQL、信号回调和后台任务，这些地方得单独补装饰器或钩子。
本文共计1166个文字，预计阅读时间需要5分钟。
在业务函数上直接增加一层装饰器，是轻量级、侵入性最小的日志记录方案。它不改变原有的逻辑，仅负责将谁、什么时间、调用了什么、传了哪些参数、结果如何等信息记录到数据库。
常见错误是把所有参数原样 json.dumps 存进去——遇到 datetime、Decimal 或自定义对象会直接报 TypeError: Object of type ... is not JSON serializable。
- 只序列化可安全落库的字段：用户ID（
current_user.id）、函数名（func.__name__）、开始时间（datetime.utcnow()）、参数快照（用repr()或白名单过滤后转字典） - 避免记录敏感参数：比如密码、token、银行卡号，装饰器里加个
exclude_keys=('password', 'token', 'card_number')参数 - 别在装饰器里做同步 DB 写入——高并发下会拖慢主流程；用
threading.Thread或异步队列（如queue.Queue配后台线程）异步落库
log_operation 必须能拿到当前用户上下文
Web 场景下，用户信息通常存在请求上下文（如 Flask 的 g.user、Django 的 request.user），但装饰器本身没 request 对象。硬写 g.user 会导致非 Web 环境（如 Celery 任务、命令行脚本）直接崩。
典型报错：RuntimeError: Working outside of application context（Flask）或 AttributeError: 'NoneType' object has no attribute 'user'（Django）。
立即学习“Python免费学习笔记（深入）”；
- 装饰器不主动取用户，改为由被装饰函数显式传入：
@log_operation(user_id=user.id) - 或统一约定一个上下文获取函数（如
get_current_user_id()），在不同环境里各自实现：Web 环境从 request 取，CLI 环境返回0或'cli'，测试环境 mock 返回固定值 - 永远不要假设
current_user是全局可用的变量——它只是框架在特定生命周期内注入的临时对象
审计日志字段设计要区分「操作行为」和「数据变更」
只记「调用了 update_profile」不够，得知道「把邮箱从 a@old.com 改成了 b@new.com」。但直接存整个 model 实例 diff 成本高、易出错。
常见翻车点：用 model_to_dict(instance) 记录前后状态，结果发现外键字段是对象而非 ID，序列化失败；或者忘记排除 updated_at 这类自动更新字段，导致每次 diff 都不一样。
- 审计表至少包含：
user_id、operation（如'update_user_email'）、target_id（被操作对象 ID）、before_data（JSON 字段，仅含关键变更前字段）、after_data（同理）、created_at - diff 逻辑收在单独函数里，比如
diff_fields(instance, fields=['email', 'phone'])，只比对业务关心的字段 - 数据库字段类型选
JSONB（PostgreSQL）或TEXT（MySQL），别用JSON类型——有些旧版 MySQL 不支持，且查询能力弱
异步写日志时注意事务与连接泄漏
用 threading.Thread(target=save_log, args=(...)) 启动新线程写 DB，容易在 SQLAlchemy 场景下触发 DetachedInstanceError 或连接池耗尽——因为主线程的 session 已关闭，子线程还在用。
更隐蔽的问题：日志写入失败 silent fail，线上完全看不到记录，审计就形同虚设。
- 子线程内新建独立 session（
sessionmaker()()），不用主线程的 session - 必须加 try/except 包裹 DB 操作，并记录错误到
logging.error；失败日志至少要包含operation、user_id、error三要素 - 别依赖线程池或全局连接池管理——简单场景用单例队列 + 定期 flush 更稳；复杂系统再上 Celery 或 Kafka
真正难的不是怎么记，是怎么保证每条关键操作都逃不过日志——漏一条，审计链就断了。最常被绕过的其实是 ORM 的 bulk 操作、raw SQL、信号回调和后台任务，这些地方得单独补装饰器或钩子。