Question 167878 in PostgreSQL Question posted in
The official documentation can be found here.

Postgresql – How to bulk insert Pydantic list using SQLAlchemy?

I am trying to insert Pydantic list to Postgres DB using SQlAlchemy but can’t figure out how to make it in a correct way.

Here is my code:

Model

class Material(Base):
    __tablename__ = 'materials'

    sap_code = Column(String, primary_key=True, nullable=False)
    sap_name = Column(String, nullable=False)
    coating = Column(String, nullable=True)
    hs_code = Column(String, nullable=True)
    quantity = Column(String, nullable=True)
    esavdo_code = Column(String, nullable=True)
    esavdo_name = Column(String, nullable=True)
    series = Column(String, nullable=True)

Schema

class MaterialBase(BaseModel):
    sap_code: str
    sap_name: str
    coating: Optional[str]
    hs_code: Optional[str]
    quantity: Optional[str]
    esavdo_code: Optional[str]
    esavdo_name: Optional[str]
    series: Optional[str]


class MaterialsList(BaseModel):
    Items: List[MaterialBase]

    class Config:
        orm_mode = True

Insert function

@router.post('/update-db', status_code=status.HTTP_200_OK, response_model=schemas.MaterialBase)
async def add_new_materials(payload: schemas.MaterialBase, db: Session = Depends(get_db)):
    db.add_all(payload)  
    db.commit()    
    db.refresh(payload)    
    return payload

Payload template

{
  "Items": [
    {
      "sap_code": "String",
      "sap_name": "String",
      "coating": "String",
      "hs_code": "String",
      "quantity": "String",
      "esavdo_code": "String",
      "esavdo_name": "String",
      "series": "String"
    },
    {
      "sap_code": "String",
      "sap_name": "String",
      "coating": "String",
      "hs_code": "String",
      "quantity": "String",
      "esavdo_code": "String",
      "esavdo_name": "String",
      "series": "String"
    }
  ]
}

Error

{
  "detail": [
    {
      "loc": [
        "body",
        "sap_code"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    },
    {
      "loc": [
        "body",
        "sap_name"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

I understand that that my schema and DB model are not matching so I have tried to parse payload manually but had mapping issues.

Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    I came up with this piece of code:

    @router.post('/update-db', status_code=status.HTTP_200_OK, response_model=schemas.MaterialsList)
async def add_new_materials(payload: schemas.MaterialsList, db: Session = Depends(get_db)):
    materials_list = payload.dict()['Items']
    for material in materials_list:
        new_material = models.Material(**material)
        db.add(new_material)
        db.commit()
        db.refresh(new_material)
    return payload

    Instead of bulk insert I decided to add one by one for the sake of simplicity and it will not effect the performance in my case


  2. 0

    It would be better to use raw queries for bulk inserting.

    db.query(
    `INSERT INTO test_table(
      name,
      age,
      mark
    )
    SELECT * FROM UNNEST (
      $1::text[],
      $2::int[],
      $3::int[]
    )`,
    [
      ["James", "Andrew"],
      [25, 27],
      [464, 745],
    ],
  )
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search