一、项目背景

“跨租户数据泄露——A 公司看到了 B 公司的所有订单！”

星云电商 SaaS 化上线第二周，安全部门在渗透测试中发现了一个严重的租户隔离漏洞。测试人员注册了两个租户账号（公司 A 和公司 B），用公司 A 的 token 调用订单列表接口，竟然看到了公司 B 的订单数据。

排查发现，问题出在"订单查询"接口——Service 层用 WHERE status = 'paid' AND created_at > '2024-01-01' 过滤，却忘记加 AND tenant_id = ?。而这个疏漏发生在 3 个 Service 和 8 个 API 接口中——没有人记得在所有查询中加入租户过滤条件。

更糟糕的是，“删除"操作也有问题。为了提高数据恢复的能力，订单中台在表上增加了 deleted_at 字段来实现"软删除”——业务删除只是标记时间戳，数据仍然保留。但开发经常忘记在查询上加 WHERE deleted_at IS NULL。某次月报统计中，包含"已删除"订单的聚合结果导致财务对账差异 $50,000。

这些问题的根因相同：全局过滤条件依赖开发者手动添加。在多租户系统中，"忘记加 tenant_id 过滤"是一个典型的全系统 bug 模式——只要有一个查询漏掉，就会导致数据泄露。SQLAlchemy 提供了两种机制来强制全局过滤：with_loader_criteria 和事件注入，可以在 ORM 层自动为所有查询注入 tenant_id = CURRENT_TENANT 和 deleted_at IS NULL 条件。

本章将通过 with_loader_criteria 和 Session 事件实现租户隔离全局过滤与软删除过滤，并实战验证"跨租户读不到"的安全性。

二、项目设计

场景：安全渗透测试结果公布后，CTO 要求"一周内修复所有租户隔离漏洞"。小胖被分配到"在所有查询中加 tenant_id 过滤"的任务——估算下来需要修改 30 多个文件。小胖绝望地来找大师求助。

小胖：“大师救命——OrderService 里有 8 个方法，每个方法里都有 select 查询，每个都要加 where(Order.tenant_id == current_tenant_id)。还要查哪些方法忘了加，这怎么排查啊？”

大师：“人工排查永远会漏。你需要一个自动的、全局的过滤机制——不需要每个开发者记得写，而是框架强制加。”

小白：“我翻文档看到 with_loader_criteria——它能在 relationship 加载时注入过滤。但普通的 select 查询呢？”

大师：“with_loader_criteria 是对 relationship 的 lazy load 和 eager load 生效的过滤器。对于你的显式 select(Order) 查询，你需要用 Session 事件——在 do_orm_execute 事件中拦截所有查询，自动追加 WHERE tenant_id = ? 和 WHERE deleted_at IS NULL。”

小胖：“技术映射：with_loader_criteria = 自动安检门（每个进出的人都要刷身份证）；Session 事件 = 天网监控（所有查询自动加上你的'身份证号'过滤条件）。”

大师：“先来看 do_orm_execute 事件的实现框架——”

@event.listens_for(Session, "do_orm_execute")
def add_tenant_filter(execute_state):
    """自动为所有 SELECT 语句注入租户和软删除过滤"""
    if execute_state.is_select and not execute_state.is_column_load:
        execute_state.statement = execute_state.statement.options(
            with_loader_criteria(
                MultiTenant,  # 所有实现 MultiTenant 的模型
                lambda cls: cls.tenant_id == current_tenant_id(),
                include_aliases=True,  # 别名也生效
            )
        )

小白：“do_orm_execute 和 before_compile 有什么区别？”

大师：“do_orm_execute 是 ORM 执行的总入口——SELECT、INSERT、UPDATE、DELETE 都会经过。before_compile 更早——在查询被编译为 SQL 字符串之前触发。对于全局过滤，do_orm_execute 足够了。如果需要修改 SQL 本身（如替换 FROM 子句），才需要 before_compile。”

小胖：“那如果某个管理员需要跨租户查看所有数据——比如报表系统——怎么绕过全局过滤？”

大师：“好问题。你需要一个’逃生门’——用 execution_options 传递标记。”

# 普通查询：自动加 tenant_id 过滤
session.execute(select(Order))

# 管理查询：跳过全局过滤
session.execute(
    select(Order).execution_options(skip_tenant_filter=True)
)

然后在事件中检查这个标记：

if not execute_state.execution_options.get("skip_tenant_filter"):
    # 注入过滤条件

小白：“技术映射：execution_options = 特殊通行证（持有者可绕过安检）。那软删除呢？”

大师：“软删除的全局过滤方式类似。但软删除还有额外的复杂性——唯一约束问题。”

class Product(Base):
    name: Mapped[str] = mapped_column(String(100), unique=True)
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

如果一个名为"iPhone"的商品被软删除（deleted_at = now()），用户想新建一个同名的"iPhone"——会触发唯一约束冲突，因为已删除的"iPhone"的 name 仍在唯一索引中。

PostgreSQL 的解决方案是部分索引：

CREATE UNIQUE INDEX idx_product_name_active
    ON products (name) WHERE deleted_at IS NULL;

这样一来，只有未删除的行才在唯一索引中——已删除的行不参与唯一性校验。"

小胖：“那 tenant_id 和 deleted_at 的全局过滤可以用同一个事件吗？”

大师：“可以，但建议拆成两个独立的事件处理器——各管各的。如果放在一起，一个出错会影响另一个。而且 skip_tenant_filter 和 skip_soft_delete_filter 也应该是独立标记。”

三、项目实战

实战目标

为订单中台实现租户隔离全局过滤与软删除过滤，验证"跨租户读不到"和"已删除数据不可见"的隔离性。

步骤一：模型声明（带租户 + 软删除）

"""ch25_multitenant.py —— 多租户与软删除模式实战"""

from sqlalchemy import (
    create_engine, String, Integer, Numeric, DateTime,
    ForeignKey, text, func, select, and_, event, Index,
    UniqueConstraint, inspect,
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship,
    Session, sessionmaker, with_loader_criteria, Bundle,
)
from sqlalchemy.sql import Select
from datetime import datetime
from typing import List, Optional, Any, TypeVar
import contextvars

# =============================================
# 租户上下文——线程安全的当前租户标识
# =============================================

current_tenant_id: contextvars.ContextVar[str] = contextvars.ContextVar(
    "current_tenant_id", default=""
)

def set_tenant(tenant_id: str):
    current_tenant_id.set(tenant_id)

def get_tenant() -> str:
    return current_tenant_id.get()

# =============================================
# 数据库引擎与基类
# =============================================

engine = create_engine(
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center",
    echo=True,
)

class Base(DeclarativeBase):
    pass

# =============================================
# Mixin：多租户 + 软删除
# =============================================

class TenantMixin:
    """租户隔离 Mixin——任何需要租户隔离的表继承此 Mixin"""
    tenant_id: Mapped[str] = mapped_column(String(50), nullable=False, index=True)

    @classmethod
    def with_tenant_filter(cls, tenant_id: str):
        """返回租户过滤条件"""
        return cls.tenant_id == tenant_id

class SoftDeleteMixin:
    """软删除 Mixin——标记删除而非物理删除"""
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True, default=None, index=True
    )

    @classmethod
    def with_active_filter(cls):
        """返回活跃记录（未删除）过滤条件"""
        return cls.deleted_at == None

class BaseModel(Base):
    """项目中所有模型共用的基类"""
    __abstract__ = True
    id: Mapped[int] = mapped_column(primary_key=True)

# =============================================
# 业务模型
# =============================================

class Tenant(BaseModel):
    __tablename__ = "mt_tenants"
    tenant_name: Mapped[str] = mapped_column(String(100), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())

class Product(BaseModel, TenantMixin, SoftDeleteMixin):
    __tablename__ = "mt_products"
    sku: Mapped[str] = mapped_column(String(30))
    title: Mapped[str] = mapped_column(String(200))
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2))
    __table_args__ = (
        # 部分唯一索引：同一租户下，未删除的 sku 唯一
        Index("idx_sku_per_tenant", "tenant_id", "sku",
              unique=True,
              postgresql_where=SoftDeleteMixin.deleted_at == None),
    )

class Order(BaseModel, TenantMixin, SoftDeleteMixin):
    __tablename__ = "mt_orders"
    order_no: Mapped[str] = mapped_column(String(32))
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2))
    status: Mapped[str] = mapped_column(String(20), default="pending")
    items: Mapped[List["OrderItem"]] = relationship(
        back_populates="order", cascade="all, delete-orphan"
    )

class OrderItem(BaseModel, TenantMixin, SoftDeleteMixin):
    __tablename__ = "mt_items"
    order_id: Mapped[int] = mapped_column(ForeignKey("mt_orders.id"))
    product_name: Mapped[str] = mapped_column(String(200))
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2))
    quantity: Mapped[int] = mapped_column(Integer)
    order: Mapped["Order"] = relationship(back_populates="items")

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

Factory = sessionmaker(bind=engine)

步骤二：do_orm_execute 全局过滤

# =============================================
# 全局过滤事件——租户 + 软删除
# =============================================

# 需要过滤的模型类型集合
TENANT_MODELS = (Product, Order, OrderItem)

@event.listens_for(Session, "do_orm_execute")
def apply_global_filters(execute_state):
    """拦截所有 ORM 查询，自动注入租户和软删除过滤"""
    if not execute_state.is_select:
        return  # 只处理 SELECT（INSERT/UPDATE/DELETE 不过滤）

    # 检查是否有"跳过"标记
    if execute_state.execution_options.get("skip_tenant_filter", False):
        tenant_filter = None
    else:
        tid = get_tenant()
        tenant_filter = tid if tid else None

    if execute_state.execution_options.get("include_deleted", False):
        soft_delete_filter = None
    else:
        soft_delete_filter = True  # 默认过滤已删除记录

    # 如果没有过滤需求，跳过
    if tenant_filter is None and not soft_delete_filter:
        return

    # 构建过滤条件
    filters = []
    if tenant_filter:
        filters.append(
            with_loader_criteria(
                TenantMixin,
                lambda cls: cls.tenant_id == tenant_filter,
                include_aliases=True,
            )
        )
    if soft_delete_filter:
        filters.append(
            with_loader_criteria(
                SoftDeleteMixin,
                lambda cls: cls.deleted_at == None,
                include_aliases=True,
            )
        )

    execute_state.statement = execute_state.statement.options(*filters)

# =============================================
# 上下文管理器：临时切换租户
# =============================================

from contextlib import contextmanager

@contextmanager
def tenant_context(tenant_id: str):
    """租户上下文——在 with 块内的所有查询自动使用此租户"""
    token = current_tenant_id.set(tenant_id)
    try:
        yield
    finally:
        current_tenant_id.reset(token)

# =============================================
# 辅助函数：软删除操作
# =============================================

def soft_delete(session: Session, obj: BaseModel):
    """执行软删除：设置 deleted_at 而不是物理删除"""
    if hasattr(obj, "deleted_at"):
        obj.deleted_at = datetime.now()
    session.add(obj)

步骤三：租户隔离验证

# =============================================
# 测试 1：租户隔离——A 看不到 B 的数据
# =============================================

print("=== 测试 1：租户隔离 ===")

with Factory() as s:
    # 创建租户 A 的数据
    with tenant_context("tenant-a"):
        p_a = Product(sku="SKU-A", title="商品A", unit_price=100, tenant_id="tenant-a")
        s.add(p_a)
        s.commit()

    # 创建租户 B 的数据
    with tenant_context("tenant-b"):
        p_b = Product(sku="SKU-B", title="商品B", unit_price=200, tenant_id="tenant-b")
        s.add(p_b)
        s.commit()

print("  [tenant-a 的视角]")
with tenant_context("tenant-a"):
    with Factory() as s:
        products = s.execute(select(Product)).scalars().all()
        print(f"  可见商品: {[p.title for p in products]}")
        assert len(products) == 1, "租户 A 应只能看到自己的商品"
        assert products[0].tenant_id == "tenant-a"

print("  [tenant-b 的视角]")
with tenant_context("tenant-b"):
    with Factory() as s:
        products = s.execute(select(Product)).scalars().all()
        print(f"  可见商品: {[p.title for p in products]}")
        assert len(products) == 1, "租户 B 应只能看到自己的商品"
        assert products[0].tenant_id == "tenant-b"

print("  [未设置租户]")
with tenant_context(""):
    with Factory() as s:
        products = s.execute(select(Product)).scalars().all()
        print(f"  可见商品（无 tenant_id 过滤）: {[p.title for p in products]}")

步骤四：软删除验证

# =============================================
# 测试 2：软删除——已删除记录不可见
# =============================================

print("\n=== 测试 2：软删除过滤 ===")

with tenant_context("tenant-a"):
    with Factory() as s:
        # 创建两个商品
        p1 = Product(sku="SKU-1", title="商品1", unit_price=100, tenant_id="tenant-a")
        p2 = Product(sku="SKU-2", title="商品2", unit_price=200, tenant_id="tenant-a")
        s.add_all([p1, p2])
        s.commit()

        # 软删除 p1
        soft_delete(s, p1)
        s.commit()

        # 查询：默认看不到已删除的
        active = s.execute(select(Product)).scalars().all()
        print(f"  活跃商品: {[p.title for p in active]}")
        assert len(active) == 1, "默认查询应过滤已删除记录"
        assert active[0].title == "商品2"

        # 管理查询：可以看到已删除的
        all_products = s.execute(
            select(Product).execution_options(include_deleted=True)
        ).scalars().all()
        print(f"  全部商品（含已删除）: {[p.title for p in all_products]}")
        assert len(all_products) == 2

# =============================================
# 测试 3：管理端跳过过滤
# =============================================

print("\n=== 测试 3：管理端跨租户查询 ===")

with Factory() as s:
    # 使用 skip_tenant_filter 跨租户查询
    all_products = s.execute(
        select(Product).execution_options(skip_tenant_filter=True, include_deleted=True)
    ).scalars().all()
    print(f"  管理端可见全部商品: {[f'{p.title}({p.tenant_id})' for p in all_products]}")
    assert len(all_products) >= 3, "管理端应能看到所有租户的数据"

步骤五：部分唯一索引验证

# =============================================
# 测试 4：部分唯一索引——同名不同租户/已删除后重新创建
# =============================================

print("\n=== 测试 4：部分唯一索引 ===")

with tenant_context("tenant-a"):
    with Factory() as s:
        # 软删除 SKU-1 后，可以创建同名的 SKU-1
        existing = s.execute(
            select(Product).where(Product.sku == "SKU-1")
            .execution_options(include_deleted=True)
        ).scalars().first()

        if existing:
            print(f"  已存在 SKU-1（deleted_at={existing.deleted_at}）")
            # 创建同 SKU 的新商品——部分唯一索引允许
            try:
                new_p = Product(sku="SKU-1", title="商品1-新版本", unit_price=150, tenant_id="tenant-a")
                s.add(new_p)
                s.commit()
                print("  创建同名 SKU 成功——部分唯一索引生效")
            except Exception as e:
                print(f"  创建同名 SKU 失败（不应出现）: {e}")

        # 验证：两个同名 SKU——一个 deleted，一个 active
        all_skus = s.execute(
            select(Product).execution_options(include_deleted=True)
            .where(Product.sku == "SKU-1")
        ).scalars().all()
        active_count = sum(1 for p in all_skus if p.deleted_at is None)
        deleted_count = sum(1 for p in all_skus if p.deleted_at is not None)
        print(f"  SKU-1: 活跃 {active_count} 条, 已删除 {deleted_count} 条")

# =============================================
# 测试 5：租户串数据检测——B 看不到 A 的数据
# =============================================

print("\n=== 测试 5：跨租户读取防护 ===")

# 在 tenant-b 上下文中尝试读取 tenant-a 的数据
with tenant_context("tenant-b"):
    with Factory() as s:
        # 尝试直接拿 tenant-a 的 ID 读取
        a_products = s.execute(
            select(Product).execution_options(skip_tenant_filter=True)
            .where(Product.tenant_id == "tenant-a")
        ).scalars().all()

        if a_products:
            a_id = a_products[0].id
            # 去掉 skip_tenant_filter 后尝试 get——应该拿不到
            p = s.get(Product, a_id)  # session.get 不走 do_orm_execute!
            print(f"  [session.get] 尝试获取租户A的商品 #{a_id}: {'获取到' if p else '被隔离（返回 None）'}")
            if p:
                print(f"    !!! 警告：tenant_id={p.tenant_id}")

可能遇到的坑及解决方法

1. session.get() 不经过 do_orm_execute

• 现象：租户 B 用 session.get(Product, id_of_tenant_a_product) 直接拿到租户 A 的数据。
• 根因：session.get() 走 Identity Map 直接内存查找，不触发 do_orm_execute。
• 解决：使用 select(Product).where(Product.id == x) 替代 session.get()，或在 session.get() 后手动检查 obj.tenant_id == current_tenant_id。或者覆写 session.get 为自定义方法。

2. with_loader_criteria 对 manual join 不生效

• 现象：select(Order).join(Product, Order.product_id == Product.id) ——手动 JOIN 后的 Product 不会被自动过滤。
• 解决：对 join 的目标表也显式写在 select(...).where(Product.tenant_id == ...) 中，或使用 with_loader_criteria 的 entity 参数指定多个模型。

3. 软删除记录的唯一约束冲突

• 现象：多次软删除和重建同名记录，某次报 unique constraint violation。
• 根因：部分唯一索引的 WHERE 条件是 deleted_at IS NULL——如果某条记录的 deleted_at 字段在软删除后又变回了 NULL（逻辑恢复），而另一条活跃记录也有相同值，就会冲突。
• 解决："恢复"操作应检查是否有活跃同名记录。常规做法是软删除后不允许恢复，或者恢复时先检查。

4. 异步环境下 contextvars 行为正常

• 现象：asyncio 中不同任务可能在同一线程中交替执行，contextvars 仍能保持正确隔离。
• 确认：Python 的 contextvars 在 asyncio 中是任务级隔离（而非线程级），所以 ContextVar 在异步代码中也是安全的。

测试验证

# tests/test_ch25_multitenant.py
import pytest
from sqlalchemy import create_engine, String, Integer, DateTime, select, event, Index
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, with_loader_criteria
from datetime import datetime
import contextvars

@pytest.fixture
def engine():
    return create_engine("sqlite:///:memory:", echo=False)

@pytest.fixture
def session(engine):
    class Base(DeclarativeBase):
        pass

    ctx = contextvars.ContextVar("tid", default="")

    class TenantMixin:
        tenant_id: Mapped[str] = mapped_column(String(50), nullable=False)
    class SoftDeleteMixin:
        deleted_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

    class Product(Base, TenantMixin, SoftDeleteMixin):
        __tablename__ = "mt_p"
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(100))
        __table_args__ = (
            Index("idx_mt_title", "tenant_id", "title", unique=True),
        )

    @event.listens_for(sessionmaker, "do_orm_execute")
    def filter(execute_state):
        if not execute_state.is_select:
            return
        tid = ctx.get()
        filters = []
        if tid and not execute_state.execution_options.get("skip_tenant_filter"):
            filters.append(with_loader_criteria(
                TenantMixin, lambda cls: cls.tenant_id == tid, include_aliases=True,
            ))
        if not execute_state.execution_options.get("include_deleted"):
            filters.append(with_loader_criteria(
                SoftDeleteMixin, lambda cls: cls.deleted_at == None, include_aliases=True,
            ))
        if filters:
            execute_state.statement = execute_state.statement.options(*filters)

    Base.metadata.create_all(engine)

    F = sessionmaker(bind=engine)
    with F() as s:
        ctx.set("tA")
        s.add(Product(title="PA", tenant_id="tA"))
        ctx.set("tB")
        s.add(Product(title="PB", tenant_id="tB"))
        s.commit()

    return F, ctx, Product

def test_tenant_isolation(session):
    F, ctx, Product = session
    with F() as s:
        ctx.set("tA")
        results = s.execute(select(Product)).scalars().all()
        assert len(results) == 1
        assert results[0].title == "PA"

    with F() as s:
        ctx.set("tB")
        results = s.execute(select(Product)).scalars().all()
        assert len(results) == 1
        assert results[0].title == "PB"

def test_soft_delete_hides_record(session):
    F, ctx, Product = session
    ctx.set("tA")
    with F() as s:
        p = s.execute(select(Product).where(Product.title == "PA")).scalars().one()
        p.deleted_at = datetime.now()
        s.commit()

    with F() as s:
        ctx.set("tA")
        active = s.execute(select(Product)).scalars().all()
        assert len(active) == 0

        all_ = s.execute(select(Product).execution_options(include_deleted=True)).scalars().all()
        assert len(all_) == 1, f"预期 1 条（含已删除），实际 {len(all_)}"

def test_admin_can_skip_filter(session):
    F, ctx, Product = session
    with F() as s:
        ctx.set("")
        results = s.execute(
            select(Product).execution_options(skip_tenant_filter=True, include_deleted=True)
        ).scalars().all()
        assert len(results) == 2

四、项目总结

过滤机制对比

机制	作用范围	优点	缺点
with_loader_criteria	relationship 的 lazy/eager load	精确、配合 options 灵活	只对 ORM 查询生效
do_orm_execute 事件	所有 ORM SELECT 查询	全局有效，开发者无感	性能微开销；需注意特殊场景
手动添加 WHERE	单个查询	最灵活	易遗漏，不够安全

租户隔离方案对比

方案	隔离强度	实现复杂度	适用场景
共享库 + tenant_id	应用层隔离	低	中小型 SaaS
独立 Schema（PostgreSQL）	数据库级隔离	中	安全要求高的场景
独立数据库	物理隔离	高	金融、医疗等强合规场景

适用场景

1. SaaS 多租户系统：所有表加 tenant_id + 全局过滤——开发时无需担心租户隔离。
2. 软删除审计追踪：所有"删除"操作改成 deleted_at 标记——数据可恢复、可审计。
3. 维护模式：系统维护时，通过全局过滤跳过待处理的数据行（如 status != 'archived'）。
4. A/B 测试：用 tenant_id 类似的模式标记"实验分组"，查询自动过滤。

不适用场景：

1. 批量数据清洗/ETL——全局过滤会干扰数据处理逻辑。
2. 跨库 JOIN 查询——with_loader_criteria 不跨数据库引擎生效。

注意事项

1. session.get() 不经过 do_orm_execute——需额外处理或统一使用 select(...).where(...)。
2. 手动 JOIN 的目标表不会被自动过滤——需要在 JOIN 后显式 where()。
3. contextvars 需要 Python 3.7+，asyncio 中自动任务隔离。
4. 不要忘记索引——tenant_id、deleted_at 需要索引，否则全局过滤条件会导致全表扫描。

常见踩坑经验

案例 1：Alembic 迁移中的测试数据绕过全局过滤

• 现象：Alembic 的 env.py 中执行了数据迁移的 session.execute(select(User))，但因为当前没有租户上下文（ContextVar 为空），查询报错或返回错误结果。
• 修复：在 Alembic 的 env.py 中使用 Core 连接（connection.execute(...)）而非 ORM Session（Core 连接不受 do_orm_execute 影响）。

案例 2：异步 Worker 中上下文丢失

• 现象：Celery/ARQ 异步任务消费订单时，ContextVar 中无租户信息——订单查询结果为空。
• 修复：在异步任务的序列化参数中显式传入 tenant_id，任务启动时调用 set_tenant(tenant_id)。

案例 3：部分唯一索引在 SQLite 上不生效

• 现象：开发用 SQLite 测试通过，上 PostgreSQL 后也通过——但 SQLite 不支持部分索引的 WHERE 子句（部分版本），测试在本地通过但在 CI 中的 SQLite 版本上失败。
• 修复：单元测试用 SQLite 验证基本行为；集成测试用 PostgreSQL 验证部分唯一索引。

思考题

1. with_loader_criteria 可以对单个关系定义过滤条件。假设 Order 有一个 items 关系（relationship("OrderItem")），如果只想加载"未删除的明细"（OrderItem.deleted_at IS NULL），应该如何配置 with_loader_criteria？这种配置和全局的软删除过滤器会冲突吗？

2. 如果你的多租户系统需要支持租户别名——例如，租户 A 在"集团 A"下，查询"集团 A"时应该包含其所有子租户的数据。全局过滤层如何支持这种树形租户结构？应该使用递归 CTE 还是预先展开所有子租户 ID 列表？各自的优劣是什么？

参考答案参见附录 E。

延伸阅读与资源

NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化）
Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Redis 实战修炼与原理进阶
Python 3实战精进：从脚本到高并发订单引擎
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
MongoDB 实战进阶与内核修炼
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析