FastAPI基于请求头差异化校验请求体的实现方案

1. 为同一接口声明不同的BaseModel类 #

直接针对每个租户的请求体结构，创建独立的Pydantic BaseModel即可，无需复杂继承，保持结构清晰：

from pydantic import BaseModel
from datetime import datetime

# Tenant1专属请求体模型
class Tenant1AccountRequest(BaseModel):
    account_id: str
    account_name: str

# Tenant2专属请求体模型
class Tenant2AccountRequest(BaseModel):
    account_id: int
    account_name: str
    created_date: datetime

每个模型严格对应租户的字段类型与必填项，后续校验时直接调用即可。

2. 基于租户头实现Pydantic校验请求体 #

通过FastAPI的Depends自定义依赖获取并验证租户头，再根据租户标识选择对应模型手动解析请求体，实现差异化校验：

from fastapi import FastAPI, Request, Depends, HTTPException

app = FastAPI()

# 自定义依赖：获取并验证租户头合法性
def validate_tenant(request: Request):
    tenant = request.headers.get("Tenant")
    if not tenant or tenant not in ["Tenant1", "Tenant2"]:
        raise HTTPException(status_code=400, detail="缺少合法的Tenant请求头")
    return tenant

@app.post("/account")
async def create_account(request: Request, tenant: str = Depends(validate_tenant)):
    # 读取原始请求体
    raw_body = await request.json()
    
    # 根据租户选择对应模型执行校验
    try:
        if tenant == "Tenant1":
            validated_data = Tenant1AccountRequest(**raw_body)
        else:
            validated_data = Tenant2AccountRequest(**raw_body)
    except Exception as e:
        # 校验失败返回标准422错误，错误信息与FastAPI默认行为一致
        raise HTTPException(status_code=422, detail=f"请求体格式错误: {str(e)}")
    
    # 后续业务逻辑示例：返回校验后的数据
    return {"status": "success", "data": validated_data.dict()}

3. 在Swagger中展示差异化内容 #

FastAPI默认Swagger不支持单接口多请求体展示，可通过多路径操作的方式，为同一/account路径创建两个逻辑独立的接口条目，分别对应不同租户：

from fastapi.openapi.models import RequestBody

# Tenant1专属接口条目
@app.post(
    "/account",
    summary="为Tenant1创建账号",
    openapi_extra={
        "parameters": [
            {
                "name": "Tenant",
                "in": "header",
                "required": True,
                "schema": {"type": "string", "enum": ["Tenant1"]}
            }
        ],
        "requestBody": RequestBody(
            content={"application/json": {"schema": Tenant1AccountRequest.model_json_schema()}}
        )
    }
)
async def create_tenant1_account(
    data: Tenant1AccountRequest,
    tenant: str = Depends(lambda req: req.headers.get("Tenant") if req.headers.get("Tenant") == "Tenant1" else None)
):
    return {"status": "success", "tenant": "Tenant1", "data": data.dict()}

# Tenant2专属接口条目
@app.post(
    "/account",
    summary="为Tenant2创建账号",
    openapi_extra={
        "parameters": [
            {
                "name": "Tenant",
                "in": "header",
                "required": True,
                "schema": {"type": "string", "enum": ["Tenant2"]}
            }
        ],
        "requestBody": RequestBody(
            content={"application/json": {"schema": Tenant2AccountRequest.model_json_schema()}}
        )
    }
)
async def create_tenant2_account(
    data: Tenant2AccountRequest,
    tenant: str = Depends(lambda req: req.headers.get("Tenant") if req.headers.get("Tenant") == "Tenant2" else None)
):
    return {"status": "success", "tenant": "Tenant2", "data": data.dict()}

在Swagger（/docs）页面中会显示两个独立的POST /account条目，分别标注租户专属的请求头要求和请求体结构，用户可直接选择对应条目进行测试。

内容的提问来源于stack exchange，提问作者Manisha Bayya