FastAPI中SQLAlchemy事件的正确使用方式及用户信息异常缓存问题求助

看起来你遇到了FastAPI结合SQLAlchemy事件时的一个典型陷阱——SQLAlchemy的事件监听是全局注册的，并非请求级别的绑定，这就是为什么之前登录用户的邮箱会被“缓存”下来，甚至在不需要用户信息的请求中也被错误使用的原因。

问题根源分析 #

你当前的代码有两个核心问题：

1. 重复注册全局事件：在get_db依赖函数内部每次请求都注册before_insert和before_update事件，而SQLAlchemy会将这些事件全局附着在Base类上，不会随请求结束而销毁。多次注册后，事件钩子会累积，且闭包捕获的current_user_email会绑定到第一次注册时的变量引用，后续请求的用户信息无法正确传递。
2. 未登录请求的事件泄漏：当使用get_db_without_current_user时，虽然没有注册新事件，但之前全局注册的事件依然生效，此时事件中引用的current_user_email还是上一次请求的旧值，导致未登录操作也被打上了之前用户的标记。

正确的解决方案：全局注册事件 + Session级传递用户信息 #

我们需要调整事件注册的时机（全局仅注册一次），并通过SQLAlchemy Session的info属性传递请求级别的用户信息，让事件能动态获取当前请求的用户。

1. 全局注册SQLAlchemy事件

将事件监听的注册移到应用启动时执行（比如在Base类定义完成后），只注册一次：

from sqlalchemy import event, func
from your_module import Base  # 替换为你实际的Base类导入路径

# 全局注册一次before_insert事件
@event.listens_for(Base, "before_insert", propagate=True)
def set_created_by(mapper, connection, target):
    # 从当前Session的info字典中获取请求级别的用户信息
    current_user_email = connection._session.info.get("current_user_email")
    if current_user_email:
        target.created_by = current_user_email
    target.created_date = func.now()

# 全局注册一次before_update事件
@event.listens_for(Base, "before_update", propagate=True)
def set_updated_audit_fields(mapper, connection, target):
    current_user_email = connection._session.info.get("current_user_email")
    if current_user_email:
        target.updated_by = current_user_email
    target.updated_date = func.now()

2. 修改DB依赖，传递用户信息到Session

在get_db依赖中，将当前用户邮箱存入Session的info字典，供全局事件获取：

from sqlalchemy.orm import Session
from your_module import SessionLocal, get_current_user_email  # 替换为实际导入

def get_db(current_user_email: str = Depends(get_current_user_email)):
    session = SessionLocal()
    try:
        # 将当前用户邮箱存入Session的info属性，供事件监听使用
        session.info["current_user_email"] = current_user_email
        yield session
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

def get_db_without_current_user():
    session = SessionLocal()
    try:
        yield session
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

方案优势说明 #

1. 全局事件仅注册一次：避免了重复注册导致的闭包变量捕获问题，事件逻辑统一维护。
2. 请求级用户信息传递：每个请求的Session都是独立的，通过session.info存储的用户信息是当前请求专属的，事件触发时能准确获取到对应请求的用户（如果有的话）。
3. 兼容无用户的场景：当使用get_db_without_current_user时，session.info中没有current_user_email，事件会自动跳过设置created_by和updated_by，符合你的需求。

额外注意事项 #

• 确保事件注册代码在应用启动时就被执行（比如在main.py中导入执行，或者在Base类定义后立即注册），避免事件未生效。
• 如果你的created_by字段是必填项，需要在无用户的场景下做特殊处理（比如设置默认值为"system"），否则会导致插入失败。

备注：内容来源于stack exchange，提问作者RheinmetallSkorpion