如何在FastAPI与SQLAlchemy中实现无域名头的多租户架构？

针对你的需求，推荐采用**共享数据库+共享Schema（单库单表带租户ID）**的模式，这种方案最简洁且扩展性强，完全不需要依赖域名头。下面是具体的实现步骤和示例：

一、租户标识的传递方式 #

不需要域名的话，推荐以下几种传递租户ID的方式（选一种最适合你的业务场景）：

• 请求头：比如X-Tenant-ID，适用于前后端分离场景，前端每次请求带上该头即可
• 路径参数：比如/api/{tenant_id}/users，适用于API网关转发或需要直观区分租户的场景
• JWT Token：把租户ID嵌入Token的Payload中，适用于需要身份认证的场景，同时传递用户和租户信息

二、核心架构实现 #

1. 数据库模型设计 #

所有租户相关的业务表都需要添加tenant_id字段，关联租户信息表：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

# 租户信息表
class Tenant(Base):
    __tablename__ = "tenants"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    # 可添加其他租户配置字段，比如过期时间、资源配额等

# 业务示例表（以用户表为例）
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    
    # 关联租户表
    tenant = relationship("Tenant")

2. SQLAlchemy自动租户过滤 #

为避免每个查询都手动添加tenant_id条件，通过SQLAlchemy会话钩子实现自动过滤：

from sqlalchemy import event
from sqlalchemy.orm import Session

def setup_tenant_filter(session: Session, tenant_id: int):
    def before_execute(conn, clauseelement, multiparams, params):
        # 仅对SELECT语句添加租户过滤，且针对包含tenant_id字段的表
        if hasattr(clauseelement, 'froms'):
            for from_obj in clauseelement.froms:
                table = from_obj.entity_namespace if hasattr(from_obj, 'entity_namespace') else from_obj
                if hasattr(table, '__table__') and 'tenant_id' in table.__table__.columns:
                    clauseelement = clauseelement.where(table.tenant_id == tenant_id)
        return clauseelement, multiparams, params

    event.listen(session, "before_execute", before_execute)

3. FastAPI依赖注入处理租户ID #

通过FastAPI依赖注入统一处理租户ID的获取与合法性验证：

from fastapi import Depends, HTTPException, Request
from sqlalchemy.orm import Session

# 数据库会话依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 获取并验证当前租户ID的依赖
def get_current_tenant_id(request: Request, db: Session = Depends(get_db)):
    # 示例：从请求头获取X-Tenant-ID
    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(status_code=400, detail="X-Tenant-ID header is required")
    
    try:
        tenant_id = int(tenant_id)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid Tenant ID")
    
    # 验证租户是否存在
    tenant = db.query(Tenant).filter(Tenant.id == tenant_id).first()
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")
    
    # 给当前会话设置租户过滤规则
    setup_tenant_filter(db, tenant_id)
    
    return tenant_id

4. 业务接口示例 #

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

# Pydantic请求模型
class UserCreate(BaseModel):
    username: str
    email: str

# 创建用户（自动关联当前租户）
@app.post("/users/")
def create_user(
    user: UserCreate, 
    tenant_id: int = Depends(get_current_tenant_id), 
    db: Session = Depends(get_db)
):
    db_user = User(**user.dict(), tenant_id=tenant_id)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 获取当前租户的所有用户（自动过滤租户数据）
@app.get("/users/")
def get_users(db: Session = Depends(get_db)):
    # 无需手动添加tenant_id过滤，会话钩子已自动处理
    users = db.query(User).all()
    return users

三、关键扩展建议 #

• 数据隔离验证：在修改/删除操作时，额外验证目标数据的tenant_id是否与当前租户一致，防止越权操作
• 索引优化：给所有带tenant_id的字段添加索引，提升过滤查询的性能
• 缓存策略：如果使用Redis等缓存，缓存键需带上tenant_id前缀，避免租户数据混淆
• 迁移脚本：使用Alembic做数据库迁移时，确保新增业务表都包含tenant_id字段
• 租户初始化：提供租户创建接口，可自动初始化租户的基础数据（比如默认角色、系统配置）

内容的提问来源于stack exchange，提问作者Who Do You think am i