本文介绍如何利用 sqlalchemy 的 `before_flush` 事件钩子，无需修改业务代码即可自动为模型（如 `project`）的 `created_by` 和 `updated_by` 字段注入当前用户标识，实现审计字段的零侵入式管理。
在构建基于 AWS API Gateway + Lambda 的 RESTful 后端时，常需为数据库记录自动记录操作人信息（如 created_by、updated_by），尤其当用户身份通过 JWT Token 解析获得时。手动在每个 add() 或属性赋值处设置这些字段不仅易出错，还严重破坏代码可维护性。SQLAlchemy 提供了优雅的解决方案：事件监听机制。
核心思路是监听 Session.before_flush 事件——该事件在 session.commit() 触发前、SQL 语句生成前执行，此时可安全访问待插入（session.new）和待更新（session.dirty）的对象集合，并按需注入用户上下文。
以下为完整实践示例（已适配您的场景）：
from sqlalchemy import event, Column, String, Integer
from sqlalchemy.orm import declarative_base, Session
import datetime
import jwt
Base = declarative_base()
class Project(Base):
__tablename__ = "project"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(256))
created_by = Column(String(256), nullable=True) # 建议设为 nullable，避免 flush 失败
updated_by = Column(String(256), nullable=True)
# ✅ 替换为您真实的 JWT 解析逻辑（注意：生产环境务必校验签名！）
def get_user_creds(token: str) -> dict:
try:
payload = jwt.decode(token, algorithms=["RS256"], options={"verify_signature": False})
return payload.get("username") or payload.get("cognito:username") or "anonymous"
except Exception:
return "anonymous"
# ? 注册全局事件监听器
@event.listens_for(Session, "before_flush")
def auto_set_audit_fields(session, flush_context, instances):
# 从请求上下文获取 JWT Token（Lambda 中通常来自 event['headers']['Authorization']）
# 此处用占位符模拟；实际需结合您的 Lambda handler 传递 token
token = "dummy.jwt.token" # ← 替换为真实 token 获取逻辑
current_user = get_user_creds(token)
# 自动填充新建对象的 created_by
for obj in session.new:
if isinstance(obj, Project):
if not obj.created_by: # 避免覆盖已显式设置的值
obj.created_by = current_user
# 自动填充已修改对象的 updated_by
for obj in session.dirty:
if isinstance(obj, Project):
# 只有当对象确实被修改时才更新 updated_by（可选增强）
state = session.identity_map.get((Project, obj.id))
if state and hasattr(state, '_modified') and state._modified:
obj.updated_by = current_user? 关键注意事项：
- Token 获取时机：before_flush 中无法直接访问 Lambda 的 event 或 context。推荐方案是：在 Lambda Handler 中解析 JWT 并将 current_user 存入 session.info（如 session.info["current_user"] = username），然后在事件处理器中读取：current_user = session.info.get("current_user", "anonymous")。
- 空值处理：建议将 created_by/updated_by 设为 nullable=True，防止因事件逻辑异常导致整个事务回滚。
- 性能影响：事件监听开销极小，但应避免在 before_flush 中执行耗时操作（如远程 HTTP 调用）。
- 扩展性：若需对多个模型启用审计字段，可将 isinstance(obj, Project) 抽象为协议检查（如判断对象是否有 created_by 属性），提升复用性。
通过此方案，您原有的业务代码完全保持简洁：
project = Project(name="project1") session.add(project) session.commit() # created_by 自动写入；后续修改属性后 commit，updated_by 自动更新
审计字段从此真正“隐形”——既保障数据可追溯性，又不增加业务逻辑负担。
