如何用SQLAlchemy批量插入Pydantic列表至PostgreSQL？

核心问题 #

你的接口接收参数和请求 payload 结构不匹配，同时未正确将 Pydantic 模型转换为 SQLAlchemy 实体实例，导致两个关键错误：

1. Pydantic 校验失败：接口期望接收单个 MaterialBase 结构（根字段包含sap_code等），但实际请求是包裹在Items数组中的列表结构，因此提示根字段缺失。
2. SQLAlchemy 操作错误：db.add_all() 需要接收 SQLAlchemy 模型实例的列表，而非 Pydantic 模型对象。

修正步骤与代码实现 #

1. 保留现有 Model 和 Schema #

你的 Material 数据库模型、MaterialBase 及 MaterialsList 序列化模型定义无需修改。

2. 修正插入接口函数 #

将接口接收参数改为 MaterialsList，并将每个 Pydantic 模型转换为 SQLAlchemy 实体后批量插入：

from typing import List
from fastapi import status, Depends, APIRouter
import schemas
from models import Material
from database import get_db
from sqlalchemy.orm import Session

router = APIRouter()

@router.post('/update-db', status_code=status.HTTP_201_CREATED, response_model=List[schemas.MaterialBase])
def add_new_materials(payload: schemas.MaterialsList, db: Session = Depends(get_db)):
    # 将每个Pydantic模型转换为SQLAlchemy实体
    material_instances = [Material(**item.dict()) for item in payload.Items]
    # 批量插入数据库
    db.add_all(material_instances)
    db.commit()
    # 刷新实例以同步数据库状态（可选，若无需返回数据库生成字段可省略）
    for instance in material_instances:
        db.refresh(instance)
    # 返回插入的物料列表
    return material_instances

3. 关键修改说明 #

• 参数类型匹配：将 payload: schemas.MaterialBase 改为 payload: schemas.MaterialsList，对应请求的Items数组结构。
• 模型转换：通过列表推导式，把每个 MaterialBase 对象的字典数据解构传入 Material 构造函数，生成 SQLAlchemy 可识别的实体实例。
• 响应模型调整：将 response_model 改为 List[schemas.MaterialBase]，匹配批量插入后的列表返回结构（也可改为schemas.MaterialsList，需同步调整返回值格式）。
• 状态码规范：批量插入成功返回201 Created，更符合 RESTful 设计规范。

验证测试 #

使用你提供的 payload 模板发送请求，接口将正确解析Items数组，完成批量插入并返回插入的物料数据。

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