You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI中使用Pydantic校验数组JSON请求并入库的实现方案

实现方案

你之前代码里写的i : EntretienModel只是Python的类型注解,仅用于IDE提示,运行时不会执行任何实际校验逻辑,等于没做校验。要实现数组元素的自动校验,直接利用FastAPI原生的Pydantic参数解析能力即可,不需要手动从Request对象里加载JSON。

第一步:修正接口入参声明,启用自动校验

FastAPI原生支持直接把请求体声明为Pydantic模型的列表,框架会自动完成:

  • 请求体JSON解析
  • 逐个校验数组内每个对象的字段是否存在、类型是否匹配
  • 自动做合理的类型转换(比如你请求里传的字符串形式的数字"60000"、"200.00"会自动转成定义的int、float类型)
  • 校验不通过时直接返回标准422错误响应,明确标注错误位置和原因,不需要你手写异常处理

第二步:优化数据库写入逻辑

你原来的循环单次commit()实现有两个明显问题:

  • 每插入一条就和数据库做一次IO交互,批量插入时性能极差
  • 没有事务回滚机制,如果中间某条插入失败,之前已经提交的记录会留在库里,产生半成功的脏数据
    优化后直接用add_all()批量加载数据,最后单次提交,同时加异常回滚逻辑即可。

完整修改后代码

修正后的接口实现

from fastapi import status, Depends, HTTPException
from sqlalchemy.exc import SQLAlchemyError
from typing import Union

# 其他你已有的导入保持不变

@app.post("/utilities/Entretien/submit", status_code=status.HTTP_200_OK)
async def create_item(operations: list[EntretienModel], db: Session = Depends(get_db)):
    # 走到这一步时,所有请求参数已经通过Pydantic校验,无需再手动判断字段合法性
    try:
        # 批量构建数据库表对象
        db_records = [TableEntretien(**op.model_dump()) for op in operations]
        # 批量添加到会话
        db.add_all(db_records)
        # 单次提交,减少数据库IO
        db.commit()
        # 如果需要拿到插入后的自增ID等字段,再统一刷新,不需要的话可以省略这步
        for record in db_records:
            db.refresh(record)
    except SQLAlchemyError as e:
        # 任何数据库异常都回滚事务,避免脏数据
        db.rollback()
        raise HTTPException(status_code=500, detail=f"数据写入失败: {str(e)}")
    
    return {"ping": "pong", "inserted_count": len(db_records)}

注意:如果你用的是Pydantic V1版本,把代码里的model_dump()替换成dict()即可。

建议调整Pydantic模型适配业务场景

你给出的示例请求里第二个元素缺少Montant字段,但当前模型里Montant是必填int类型,会直接触发校验错误。如果业务上该字段允许为空,建议调整模型定义,还可以加字段值规则校验:

from pydantic import BaseModel, Field

class EntretienModel(BaseModel):
    Code: str
    Agent: str
    Kilometrage: int = Field(ge=0, description="里程数不能为负")
    Operation: str
    Piece_Consomable: str
    Quantité: int = Field(gt=0, description="消耗品数量必须大于0")
    UnitPrice: float = Field(ge=0, description="单价不能为负")
    Montant: Union[int, None] = None
    Mainoeuvre: Union[int, None] = None

优化点说明

  • 校验逻辑完全由框架托管,不需要手写任何字段判断、类型转换代码,错误返回格式统一
  • 批量写入+单次提交的性能比循环单次提交高几倍到几十倍,数据量越大差距越明显
  • 事务回滚机制保证要么所有数据全部插入成功,要么全部失败,不会出现部分写入的异常情况
  • 代码量更少,逻辑更清晰,后续维护成本更低

内容的提问来源于stack exchange,提问作者MoRoPoPi

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.27 03:18:14