验证数据库中存在外键。 FastAPI + Pydantic

Question

我有一个带有另一个表的外键的数据库表模型：

class Vehicle(Base):
    """
    Vehicles db table.
    """
    __tablename__ = "vehicle"

    id: Mapped[int] = mapped_column(primary_key=True)
    license_plate: Mapped[str] = mapped_column(String(20))
    garage_number: Mapped[Optional[str]] = mapped_column(String(20))
    brand: Mapped[str] = mapped_column(String(255))
    base_location: Mapped[Optional[WKBElement]] = mapped_column(
        Geometry(geometry_type="POINT", srid=4326, spatial_index=True)
    )

    vehicle_type = relationship("VehicleType", back_populates="vehicles")
    vehicle_type_id: Mapped[int] = mapped_column(
        ForeignKey("vehicle_type.id", ondelete="RESTRICT")
    )

Pydantic 模型：

class VehicleCreate(BaseModel):
    """
    Base vehicle schema.
    """
    license_plate: str = Field(max_length=20)
    garage_number: str = Field(max_length=20)
    brand: str = Field(max_length=255)
    vehicle_type_id: int

发帖方式：

@router.post("/", response_model=VehicleRead, status_code=status.HTTP_201_CREATED)
def create_vehicle(vehicle_in: VehicleCreate, db_session: Session = Depends(get_db)):
    """
    Create a new vehicle.
    """
    return service.create_vehicle(db_session=db_session, vehicle_in=vehicle_in)

服务：

def create_vehicle(db_session: Session, vehicle_in: VehicleCreate) -> Vehicle:
    """
    Creates a vehicle.
    """
    vehicle = Vehicle(**vehicle_in.model_dump())
    db_session.add(vehicle)
    db_session.commit()
    db_session.refresh(vehicle)
    return vehicle

所以主要问题是我必须验证 vehicle_type_id 字段值是否存在 db.如果我不验证这一点，我会得到 IntegrityError。

sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table "vehicle" violates foreign key constraint "vehicle_vehicle_type_id_fkey"
DETAIL:  Key (vehicle_type_id)=(0) is not present in table "vehicle_type".

但是在哪一层呢？
我无法在 pydantic 模型中执行此操作，因为我无法访问其中的session。
我可以在视图中执行此操作并返回 JSONResponse 以及相应的错误和 422 状态。但我认为这不是地方。
我认为创建函数内的 service 模块中最好的地方，但我不知道如何在模型外引发 pydantic ValidationError 。 Django Rest框架在序列化器中具有类似的功能，即PrimaryKeyRelatedField。

Answer 1

有很多方法可以实现这一目标。

一种方法是向数据添加抽象层并使用 EAFP 原则：

# vehicle_repository.py

def create_vehicle(db_session: Session, vehicle_in: VehicleCreate) -> Vehicle:
    vehicle = Vehicle(**vehicle_in.model_dump())
    try:
        db_session.add(vehicle)
        db_session.commit()
        db_session.refresh(vehicle)
        return vehicle
    except sqlalchemy.exc.IntegrityError:
        # do what you need when it happens.
        # possibly raise some business error

第二种方法将在您的服务中添加验证，以在保存之前检查该类型是否存在（LBYL）。

# vehicle_service.py

def create_vehicle(db_session: Session, vehicle_in: VehicleCreate) -> Vehicle:
    type_exists = vehicle_repository.find_type_by_id(vehicle_in.vehicle_type_id)
    if type_exists:
        vehicle = Vehicle(**vehicle_in.model_dump())
        db_session.add(vehicle)
        db_session.commit()
        db_session.refresh(vehicle)
        return vehicle
    else:
        # raise some error

如果您想根据数据库中的值/选择验证车辆类型 ID，您可以尝试使用缓存的Validation Context。