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

huangapple
go评论110阅读模式
英文:

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

  1. class Material(Base):
  2.     __tablename__ = 'materials'
  4.     sap_code = Column(String, primary_key=True, nullable=False)
  5.     sap_name = Column(String, nullable=False)
  6.     coating = Column(String, nullable=True)
  7.     hs_code = Column(String, nullable=True)
  8.     quantity = Column(String, nullable=True)
  9.     esavdo_code = Column(String, nullable=True)
  10.     esavdo_name = Column(String, nullable=True)
  11.     series = Column(String, nullable=True)

Schema

  1. class MaterialBase(BaseModel):
  2.     sap_code: str
  3.     sap_name: str
  4.     coating: Optional[str]
  5.     hs_code: Optional[str]
  6.     quantity: Optional[str]
  7.     esavdo_code: Optional[str]
  8.     esavdo_name: Optional[str]
  9.     series: Optional[str]
  12. class MaterialsList(BaseModel):
  13.     Items: List[MaterialBase]
  15.     class Config:
  16.         orm_mode = True

Insert function

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

Payload template

  1. {
  2.   "Items": [
  3.     {
  4.       "sap_code": "String",
  5.       "sap_name": "String",
  6.       "coating": "String",
  7.       "hs_code": "String",
  8.       "quantity": "String",
  9.       "esavdo_code": "String",
  10.       "esavdo_name": "String",
  11.       "series": "String"
  12.     },
  13.     {
  14.       "sap_code": "String",
  15.       "sap_name": "String",
  16.       "coating": "String",
  17.       "hs_code": "String",
  18.       "quantity": "String",
  19.       "esavdo_code": "String",
  20.       "esavdo_name": "String",
  21.       "series": "String"
  22.     }
  23.   ]
  24. }

Error

  1. {
  2.   "detail": [
  3.     {
  4.       "loc": [
  5.         "body",
  6.         "sap_code"
  7.       ],
  8.       "msg": "field required",
  9.       "type": "value_error.missing"
  10.     },
  11.     {
  12.       "loc": [
  13.         "body",
  14.         "sap_name"
  15.       ],
  16.       "msg": "field required",
  17.       "type": "value_error.missing"
  18.     }
  19.   ]
  20. }

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

答案1

得分: 0

以下是翻译好的内容：

"建议使用原始查询进行批量插入。

db.query(
INSERT INTO test_table( name, age, mark ) SELECT * FROM UNNEST ( $1::text[], $2::int[], $3::int[] ),
[
["James", "Andrew"],
[25, 27],
[464, 745],
],
)"

英文:

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

  1. db.query(
  2.     `INSERT INTO test_table(
  3.       name,
  4.       age,
  5.       mark
  6.     )
  7.     SELECT * FROM UNNEST (
  8.       $1::text[],
  9.       $2::int[],
  10.       $3::int[]
  11.     )`,
  12.     [
  13.       ["James", "Andrew"],
  14.       [25, 27],
  15.       [464, 745],
  16.     ],
  17.   )

答案2

得分: 0

I came up with this piece of code:

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

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

英文:

I came up with this piece of code:

  1. @router.post('/update-db', status_code=status.HTTP_200_OK, response_model=schemas.MaterialsList)
  2. async def add_new_materials(payload: schemas.MaterialsList, db: Session = Depends(get_db)):
  3.     materials_list = payload.dict()['Items']
  4.     for material in materials_list:
  5.         new_material = models.Material(**material)
  6.         db.add(new_material)
  7.         db.commit()
  8.         db.refresh(new_material)
  9.     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

huangapple
  • 本文由 发表于 2023年5月11日 16:50:13
  • 转载请务必保留本文链接：https://go.coder-hub.com/76225772.html
  • fastapi
  • postgresql
  • pydantic
  • python
  • sqlalchemy
匿名

发表评论

匿名网友
#

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定