如何自动用客户端用户信息填充SQLAlchemy模型的created_by/updated_by字段

实现思路 #

利用SQLAlchemy的事件监听机制，在数据插入、更新前自动从JWT解析的用户信息填充对应字段，无需手动设置。

步骤1：完善模型定义 #

首先给Project模型添加主键（SQLAlchemy要求实体必须有主键）：

class Project(Base):
    __tablename__ = "project"
    id = Column(Integer, primary_key=True, autoincrement=True)  # 新增主键
    name = Column(String(256))
    created_by = Column(String(256))
    updated_by = Column(String(256))

步骤2：编写全局事件监听器 #

给所有继承Base的模型绑定before_insert和before_update事件，自动填充字段：

from sqlalchemy import event

# 插入数据前填充created_by和updated_by
@event.listens_for(Base, 'before_insert', propagate=True)
def auto_fill_created_by(mapper, connection, target):
    current_user = connection.session.info.get("current_user")
    if current_user:
        target.created_by = current_user
        target.updated_by = current_user

# 更新数据前更新updated_by
@event.listens_for(Base, 'before_update', propagate=True)
def auto_fill_updated_by(mapper, connection, target):
    current_user = connection.session.info.get("current_user")
    if current_user:
        target.updated_by = current_user

• propagate=True确保所有Base的子类模型都能继承该事件逻辑。
• 通过connection.session.info获取当前请求的用户信息，避免并发冲突。

步骤3：在Lambda请求中传递用户信息 #

在Lambda处理函数中，解析JWT令牌获取用户名，并存入SQLAlchemy Session的info属性：

def lambda_handler(event, context):
    # 从请求头提取JWT令牌（格式：Bearer <token>）
    auth_header = event.get("headers", {}).get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        return {"statusCode": 401, "body": "Unauthorized"}
    
    token = auth_header.split(" ")[1]
    user_creds = get_user_creds(token)
    current_user = user_creds.get("username")  # 根据实际JWT字段调整，比如sub
    
    # 创建Session并注入当前用户
    session = create_session()  # 替换为你的Session创建逻辑
    session.info["current_user"] = current_user
    
    try:
        # 业务逻辑示例：创建Project
        project = Project(name="project1")
        session.add(project)
        session.commit()
        return {"statusCode": 200, "body": "Project created"}
    finally:
        session.close()

关键注意事项 #

1. 确保每个请求的Session是独立的，避免用户信息串扰（Lambda中使用常规Session或ScopedSession均可）。
2. 根据实际JWT payload调整current_user的提取字段（比如sub、email等）。
3. 若业务要求字段不能为空，需在无用户信息时抛出异常或拦截请求，避免空值写入数据库。

内容的提问来源于stack exchange，提问作者Wajih Katrou