SQLAlchemy多租户场景下全局拦截动态修改查询/插入语句咨询

你之前用load事件改查询不生效的核心原因是触发时机完全不对：load事件是SQL已经执行完成、结果集已经映射成ORM实例之后才会触发，这时候再修改context.query对象，根本不会影响已经跑完的数据库请求，自然看不到效果。

要实现全局拦截查询、增删改操作自动追加company_id过滤，不需要给每个模型单独绑定事件，用SQLAlchemy提供的全局ORM执行事件+基类事件传播就能实现，具体方案如下：

核心实现步骤 #

1. 定义多租户模型统一基类 #

所有需要做租户隔离的模型统一继承带company_id字段的基类，方便后续统一识别、批量处理：

from contextvars import ContextVar
from sqlalchemy import Column, Integer, event
from sqlalchemy.orm import declarative_base, Session

# 协程/线程安全的上下文变量，存当前请求对应的租户ID，web场景下不会串数据
current_company_id = ContextVar("current_company_id", default=None)

# 多租户模型基类，统一带租户ID字段
class TenantModelBase:
    company_id = Column(Integer, index=True, nullable=False)

Base = declarative_base(cls=TenantModelBase)

# 业务模型直接继承Base即可，自动带company_id字段
class Settings(Base):
    __tablename__ = "settings"
    id = Column(Integer, primary_key=True)
    # 其他业务字段...

2. 全局拦截查询、更新、删除语句，自动追加租户过滤 #

绑定Session级别的do_orm_execute事件，这个事件会在所有ORM语句执行前触发，全局只需要绑定一次，就能拦截所有模型的SELECT、UPDATE、DELETE操作：

@event.listens_for(Session, "do_orm_execute")
def append_tenant_filter(exec_state):
    # 跳过关联字段的懒加载场景，这类查询会自动继承主查询的过滤条件，不需要重复加
    if exec_state.is_column_load or exec_state.is_relationship_load:
        return
    # 支持手动标记跳过租户过滤的场景，比如平台管理员跨租户操作
    if exec_state.execution_options.get("skip_tenant_filter", False):
        return

    tenant_id = current_company_id.get()
    # 未设置租户ID的处理逻辑可以根据业务调整，比如直接抛错禁止无租户查询
    if tenant_id is None:
        return

    # 处理普通SELECT查询
    if exec_state.is_select:
        stmt = exec_state.statement
        # 遍历查询涉及的所有实体，给租户模型追加过滤条件
        for desc in stmt.column_descriptions:
            model = desc["type"]
            if hasattr(model, "company_id"):
                stmt = stmt.where(model.company_id == tenant_id)
        exec_state.statement = stmt
    # 处理批量UPDATE、DELETE操作
    elif exec_state.is_update or exec_state.is_delete:
        model = exec_state.entity_description["type"]
        if hasattr(model, "company_id"):
            exec_state.statement = exec_state.statement.where(
                model.company_id == tenant_id
            )

3. 拦截插入操作，自动注入租户ID #

绑定基类的before_insert事件，开启propagate=True之后，所有继承该基类的子类模型都会自动触发这个事件，不需要逐个配置：

@event.listens_for(Base, "before_insert", propagate=True)
def inject_tenant_id(mapper, connection, target):
    tenant_id = current_company_id.get()
    if tenant_id and hasattr(target, "company_id") and target.company_id is None:
        target.company_id = tenant_id

使用方式 #

• 在web请求的入口中间件里，解析当前登录用户的company_id，设置到上下文变量中：

  # 请求进入时设置
  token = current_company_id.set(当前登录用户所属公司ID)
  try:
      # 处理请求逻辑
      ...
  finally:
      # 请求结束后重置，避免上下文污染
      current_company_id.reset(token)
  

• 所有正常的ORM查询、新增、修改、删除操作都会自动带上company_id过滤，不需要业务代码手动写条件。
• 特殊场景需要跨租户操作时，给查询加执行选项跳过过滤即可：

  # 跨租户查询所有配置
  db.query(Settings).execution_options(skip_tenant_filter=True).all()
  

注意事项 #

• 不要在load、refresh这类实例加载后的事件里尝试修改查询语句，这类事件触发时SQL已经执行完毕，修改语句不会生效。
• 基于ContextVar存储当前租户ID是协程/线程安全的，在FastAPI、Flask等主流web框架下使用不会出现并发请求租户ID串扰的问题。
• 关联表的懒加载、联合查询场景会自动继承主查询的租户过滤条件，不会出现跨租户泄露关联数据的问题。

内容的提问来源于stack exchange，提问作者Lewis Morris