本文介绍如何利用 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
? 关键注意事项：
