FastAPI+Pydantic中如何验证外键在数据库中存在？

方法一：服务层手动验证并抛出ValidationError #

直接在服务层的创建逻辑里先校验外键对应的记录是否存在，不存在就手动构造Pydantic的ValidationError抛出，FastAPI会自动将其转换为标准的422参数错误响应。

代码示例 #

from pydantic import ValidationError
from sqlalchemy.orm import Session
from your_app.models import Vehicle, VehicleType
from your_app.schemas import VehicleCreate

def create_vehicle(db: Session, vehicle_data: VehicleCreate):
    # 验证vehicle_type_id对应的记录是否存在
    vehicle_type = db.query(VehicleType).filter(VehicleType.id == vehicle_data.vehicle_type_id).first()
    if not vehicle_type:
        raise ValidationError(
            errors=[
                {
                    "loc": ("vehicle_type_id",),
                    "msg": "无效的车辆类型ID：该ID不存在",
                    "type": "value_error.not_exist",
                }
            ],
            model=VehicleCreate,
        )
    # 验证通过后创建Vehicle记录
    db_vehicle = Vehicle(**vehicle_data.dict())
    db.add(db_vehicle)
    db.commit()
    db.refresh(db_vehicle)
    return db_vehicle

方法二：自定义带数据库验证的Pydantic字段 #

如果需要复用这类外键验证逻辑，可以自定义一个Pydantic字段类型，在验证时传入数据库Session。因为Pydantic模型本身无法持有Session，需在服务层调用时主动传入上下文。

自定义字段示例 #

from pydantic import GetCoreSchemaHandler, ValidationInfo
from pydantic_core import core_schema
from sqlalchemy.orm import Session
from your_app.models import VehicleType

class ExistingVehicleTypeId(int):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        return core_schema.no_info_after_validator_function(
            cls.validate,
            core_schema.int_schema(),
        )
    
    @classmethod
    def validate(cls, v: int, info: ValidationInfo) -> int:
        # 从上下文获取数据库Session
        db: Session = info.context.get("db")
        if not db:
            raise ValueError("验证需要传入数据库Session")
        if not db.query(VehicleType).filter(VehicleType.id == v).first():
            raise ValueError(f"ID为{v}的车辆类型不存在")
        return v

在Schema中使用并在服务层传入Session #

from pydantic import BaseModel

class VehicleCreate(BaseModel):
    vehicle_type_id: ExistingVehicleTypeId
    # 其他字段...

# 服务层调用逻辑
def create_vehicle(db: Session, vehicle_data: dict):
    try:
        validated_data = VehicleCreate(**vehicle_data, context={"db": db})
    except ValidationError as e:
        raise e
    # 后续创建逻辑...

方法三：利用FastAPI依赖项抽离验证逻辑 #

如果不想在服务层直接处理Pydantic验证，可以把外键验证做成独立依赖项，在接口层提前完成校验——这是把验证逻辑抽离到独立模块，而非直接写在视图函数里，同样符合分层设计思想。

依赖项示例 #

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from your_app.db import get_db
from your_app.models import VehicleType

def validate_vehicle_type_exists(vehicle_type_id: int, db: Session = Depends(get_db)):
    if not db.query(VehicleType).filter(VehicleType.id == vehicle_type_id).first():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail={"vehicle_type_id": "无效ID：该车辆类型不存在"}
        )
    return vehicle_type_id

在接口中使用 #

from fastapi import APIRouter, Depends
from your_app.schemas import VehicleCreate
from your_app.services import create_vehicle

router = APIRouter()

@router.post("/vehicles/")
def create_vehicle_endpoint(
    vehicle_data: VehicleCreate,
    _: int = Depends(validate_vehicle_type_exists),
    db: Session = Depends(get_db)
):
    # 此处vehicle_type_id已通过验证，直接调用服务层
    return create_vehicle(db, vehicle_data)

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