FastAPI能否创建依赖链？含I/O操作的JSON验证实现咨询

FastAPI完全支持创建依赖链，这正是处理需要数据库连接或其他IO操作验证的标准方案。你给出的代码里Depends(f1, f2, f3)写法不对——Depends()只能接收单个依赖，要实现多步验证，有以下几种正确方式：

方式一：串联式依赖链（依次执行验证） #

让每个验证依赖前一步的输出，形成链条，只要某一步验证失败就直接抛出错误终止流程：

from __future__ import annotations
from fastapi import FastAPI, Body, Depends
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel

app = FastAPI()

class ImportModel(BaseModel):
    id: int
    text: str | None

# 模拟需要IO的检查函数，比如查数据库、调用外部API
def extra_check1(req: ImportModel) -> bool:
    # 示例：检查ID是否合法
    return req.id != 0

def extra_check2(req: ImportModel) -> bool:
    # 示例：检查文本长度
    return req.text is None or len(req.text) <= 100

def extra_check3(req: ImportModel) -> bool:
    # 示例：额外ID校验
    return req.id > 0

# 第一步验证：基础模型解析 + extra_check1
def validate_step1(req: ImportModel = Body()) -> ImportModel:
    if not extra_check1(req):
        # 按FastAPI标准格式抛出验证错误
        raise RequestValidationError([
            {"loc": ["id"], "msg": "ID不符合业务规则（step1）", "type": "value_error"}
        ])
    return req

# 第二步验证：依赖step1的结果 + extra_check2
def validate_step2(req: ImportModel = Depends(validate_step1)) -> ImportModel:
    if not extra_check2(req):
        raise RequestValidationError([
            {"loc": ["text"], "msg": "文本长度超出限制（step2）", "type": "value_error"}
        ])
    return req

# 第三步验证：依赖step2的结果 + extra_check3
def validate_step3(req: ImportModel = Depends(validate_step2)) -> ImportModel:
    if not extra_check3(req):
        raise RequestValidationError([
            {"loc": ["id"], "msg": "ID必须大于0（step3）", "type": "value_error"}
        ])
    return req

@app.post('/import')
def import_smth(req: ImportModel = Depends(validate_step3)):
    return req

方式二：组合式验证（单依赖整合所有检查） #

如果验证逻辑不需要分步复用，也可以把所有检查放到一个依赖函数里，更简洁：

def validate_all(req: ImportModel = Body()) -> ImportModel:
    if not extra_check1(req):
        raise RequestValidationError([{"loc": ["id"], "msg": "ID不符合规则（f1）", "type": "value_error"}])
    if not extra_check2(req):
        raise RequestValidationError([{"loc": ["text"], "msg": "文本过长（f2）", "type": "value_error"}])
    if not extra_check3(req):
        raise RequestValidationError([{"loc": ["id"], "msg": "ID必须大于0（f3）", "type": "value_error"}])
    return req

@app.post('/import')
def import_smth(req: ImportModel = Depends(validate_all)):
    return req

方式三：多独立依赖（并行验证） #

如果每个验证逻辑相互独立，可以在路由里声明多个Depends参数，每个依赖都会单独执行：

def validate_f1(req: ImportModel = Body()) -> ImportModel:
    if not extra_check1(req):
        raise RequestValidationError([{"loc": ["id"], "msg": "ID不符合规则（f1）", "type": "value_error"}])
    return req

def validate_f2(req: ImportModel = Body()) -> ImportModel:
    if not extra_check2(req):
        raise RequestValidationError([{"loc": ["text"], "msg": "文本过长（f2）", "type": "value_error"}])
    return req

@app.post('/import')
def import_smth(
    _: ImportModel = Depends(validate_f1),
    req: ImportModel = Depends(validate_f2)
):
    return req

处理数据库等IO依赖 #

如果验证需要数据库连接，只需把数据库会话作为额外依赖注入到验证函数即可：

from sqlalchemy.orm import Session
# 假设你已经定义了数据库会话依赖
from your_module import get_db

def extra_check1(req: ImportModel, db: Session = Depends(get_db)) -> bool:
    # 实际数据库查询：检查ID是否存在
    return db.query(YourModel).filter(YourModel.id == req.id).first() is not None

def validate_f1(req: ImportModel = Body(), db: Session = Depends(get_db)) -> ImportModel:
    if not extra_check1(req, db):
        raise RequestValidationError([{"loc": ["id"], "msg": "该ID不存在于数据库", "type": "value_error"}])
    return req

关键注意事项 #

• 抛出RequestValidationError时，要遵循FastAPI的错误格式（包含loc、msg、type的字典列表），这样会自动返回标准的422错误响应。
• 依赖链不仅用于验证，还可以复用逻辑（比如多个路由都需要相同的IO检查）。

内容的提问来源于stack exchange，提问作者Code Is my life