FastapiAdmin/backend/docs/数据隔离快速参考.md

多租户数据隔离快速参考

🎯 核心原则

三层隔离架构

系统层 (System) → 租户层 (Tenant) → 客户层 (Customer)

隔离字段决策表

问题	tenant_id	customer_id
是租户表本身?	❌	❌
是客户表?	✅ 必填	❌
是组织架构(部门/角色/岗位)?	✅ 必填	❌
是用户表?	✅ 必填	✅ 可选
是配置类(菜单/字典/参数)?	✅ 可选	❌
是业务数据?	✅ 必填	✅ 根据业务
是日志/通知?	✅ 必填	✅ 根据操作人

📊 数据权限 (data_scope)

值	名称	SQL WHERE 条件
1	仅本人	created_id = current_user.id
2	本部门	user.dept_id = current_user.dept_id
3	本部门及以下	dept.tree_path LIKE 'current%'
4	全部数据	tenant_id = current_user.tenant_id
5	自定义	dept_id IN (role_depts)

客户用户特殊规则: 无论什么权限,都额外加 customer_id = current_user.customer_id

🔧 代码模板

1. 租户级业务表

class ProductModel(ModelMixin):
    """产品表 - 租户级"""
    __tablename__ = "business_product"
    
    name: Mapped[str] = mapped_column(String(100), comment="产品名称")
    
    # 数据隔离
    tenant_id: Mapped[int] = mapped_column(
        Integer, 
        ForeignKey("system_tenant.id", ondelete="CASCADE"), 
        nullable=False, 
        index=True
    )
    created_id: Mapped[int | None] = mapped_column(
        Integer, 
        ForeignKey("system_user.id", ondelete="SET NULL"), 
        nullable=True
    )

2. 客户级业务表

class OrderModel(ModelMixin):
    """订单表 - 客户级"""
    __tablename__ = "business_order"
    
    order_no: Mapped[str] = mapped_column(String(50), comment="订单号")
    
    # 数据隔离
    tenant_id: Mapped[int] = mapped_column(
        Integer, 
        ForeignKey("system_tenant.id", ondelete="CASCADE"), 
        nullable=False, 
        index=True
    )
    customer_id: Mapped[int | None] = mapped_column(
        Integer, 
        ForeignKey("system_customer.id", ondelete="CASCADE"), 
        nullable=True,  # 可选,客户订单时必填
        index=True
    )
    created_id: Mapped[int | None] = mapped_column(
        Integer, 
        ForeignKey("system_user.id", ondelete="SET NULL"), 
        nullable=True
    )

3. 查询过滤器

class DataPermissionFilter:
    """通用数据权限过滤器"""
    
    @staticmethod
    def apply(query: Select, model, user: UserModel) -> Select:
        # 1. 租户隔离 (必须)
        query = query.where(model.tenant_id == user.tenant_id)
        
        # 2. 客户隔离 (如果是客户用户)
        if user.customer_id:
            query = query.where(model.customer_id == user.customer_id)
        
        # 3. 数据权限
        max_scope = max(int(r.data_scope) for r in user.roles)
        
        if max_scope == 1:  # 仅本人
            query = query.where(model.created_id == user.id)
        
        elif max_scope == 2:  # 本部门
            query = query.where(
                model.created_id.in_(
                    select(UserModel.id)
                    .where(UserModel.dept_id == user.dept_id)
                )
            )
        
        elif max_scope == 3:  # 本部门及以下
            query = query.where(
                model.created_id.in_(
                    select(UserModel.id)
                    .join(DeptModel)
                    .where(DeptModel.tree_path.like(f'{user.dept.tree_path}%'))
                )
            )
        
        # 4 和 5 省略...
        
        return query

4. 创建记录

async def create_order(data: dict, current_user: UserModel):
    new_order = OrderModel(
        **data,
        tenant_id=current_user.tenant_id,  # 必须
        customer_id=current_user.customer_id,  # 如果是客户用户
        created_id=current_user.id,
        updated_id=current_user.id
    )
    session.add(new_order)
    await session.commit()

⚠️ 常见错误

❌ 错误1: 缺少租户过滤

# 危险! 可能查到其他租户的数据
query = select(Model).where(Model.name == 'xxx')

✅ 正确:

query = (
    select(Model)
    .where(Model.tenant_id == current_user.tenant_id)
    .where(Model.name == 'xxx')
)

❌ 错误2: 客户用户缺少customer_id过滤

# 客户用户可能看到其他客户的数据
query = select(Model).where(Model.tenant_id == current_user.tenant_id)

✅ 正确:

query = select(Model).where(Model.tenant_id == current_user.tenant_id)
if current_user.customer_id:
    query = query.where(Model.customer_id == current_user.customer_id)

❌ 错误3: 给不该有的表加customer_id

# 错误: 部门不属于客户
class DeptModel(ModelMixin):
    customer_id: Mapped[int]  # ❌

✅ 正确:

class DeptModel(ModelMixin):
    tenant_id: Mapped[int]  # ✅ 只需要tenant_id

❌ 错误4: 租户表加tenant_id

# 错误: 租户不属于租户
class TenantModel(ModelMixin):
    tenant_id: Mapped[int]  # ❌ 循环引用

✅ 正确:

class TenantModel(ModelMixin):
    # 不需要tenant_id ✅
    created_id: Mapped[int | None]  # 只需要审计字段

📝 检查清单

创建新表时,请检查:

• 是否需要 tenant_id? (几乎所有业务表都需要)
• 是否需要 customer_id? (只有部分表需要)
• 是否需要 created_id/updated_id? (推荐加上,用于审计)
• 是否创建了必要的索引?
  • idx_tenant_id
  • idx_customer_id (如果有)
  • idx_created_id (如果有)
• 关联关系是否正确设置了 foreign_keys?
• 是否在 TYPE_CHECKING 下导入相关模型?

查询时,请检查:

• 是否添加了 tenant_id 过滤?
• 如果当前用户是客户用户,是否添加了 customer_id 过滤?
• 是否根据角色的 data_scope 添加了数据权限过滤?
• 是否使用了数据权限过滤器中间件?

创建/更新记录时,请检查:

• 是否设置了 tenant_id?
• 是否根据当前用户设置了 customer_id?
• 是否设置了 created_id/updated_id?

🚀 性能优化

必须创建的索引

-- 单列索引
CREATE INDEX idx_tenant_id ON table_name(tenant_id);
CREATE INDEX idx_customer_id ON table_name(customer_id);
CREATE INDEX idx_created_id ON table_name(created_id);

-- 联合索引 (提升查询性能)
CREATE INDEX idx_tenant_customer ON table_name(tenant_id, customer_id);
CREATE INDEX idx_tenant_status ON table_name(tenant_id, status);

-- 树形路径索引 (本部门及以下查询)
CREATE INDEX idx_tree_path ON system_dept(tree_path);

分区表 (大数据量场景)

-- 按租户分区
CREATE TABLE orders (
    id BIGINT,
    tenant_id INT,
    ...
) PARTITION BY HASH(tenant_id) PARTITIONS 10;

📚 相关文档

详细设计请参考: 多租户数据隔离设计方案.md