如何使用 sqlalchemy add_all 在 Fast API 中批量保存答案

【问题标题】：How to bulk save in Fast API using sqlalchemy add_all如何使用 sqlalchemy add_all 在 Fast API 中批量保存
【发布时间】：2020-11-15 20:49:20
【问题描述】：

我正在尝试将患者批量添加到数据库中，但遇到了错误。目标是从请求正文中读取数据，截断表中的数据并添加新数据。谁能告诉我我做错了什么？

代码

schemas.py

from pydantic import BaseModel
from typing import Optional

class PatientBase(BaseModel):
    ticket_id: str
    patient_name: Optional[str] = None



class PatientInDb(PatientBase):
    patient_id : str
    institute :str

    class Config:
        orm_mode = True

crud.py

from typing import List
from sqlalchemy.orm import Session

def create_patients(db: Session, patients: List[schemas.PatientInDb] ):
    num_of_deleted_rows = db.query(models.Patient).delete()

    db.add_all(patients)
    db.commit()
    return db.query(models.Patient).count()

患者.py


@router.post("/patients")
async def post_patients(
    patients : List[schemas.PatientInDb],
    db: Session = Depends(get_db),
    
):
    patients_count = crud.create_patients(db, patients)
    return {
        "message":f"New {patients_count} patients created."
    }

错误

  File ".\app\api\v1\patients.py", line 45, in post_patients
    patients_count = crud.create_patients(db, patients)
  File ".\app\crud.py", line 13, in create_patients
    db.add_all(patients)
  File "c:\users\convergytics\miniconda3\envs\test\lib\site-packages\sqlalchemy\orm\session.py", line 2016, in add_all
    for instance in instances:
  File "c:\users\convergytics\miniconda3\envs\test\lib\typing.py", line 682, in inner
    return func(*args, **kwds)
  File "c:\users\convergytics\miniconda3\envs\test\lib\typing.py", line 1107, in __getitem__
    params = tuple(_type_check(p, msg) for p in params)
  File "c:\users\convergytics\miniconda3\envs\test\lib\typing.py", line 1107, in <genexpr>
    params = tuple(_type_check(p, msg) for p in params)
  File "c:\users\convergytics\miniconda3\envs\test\lib\typing.py", line 374, in _type_check
    raise TypeError(msg + " Got %.100r." % (arg,))
TypeError: Parameters to generic types must be types. Got 0.

【问题讨论】：

在没有任何where 子句的情况下删除请求中的所有患者听起来是个坏主意。最好的情况是，您将在 SQL 中有很多死垃圾行和一个用于持续删除和添加相同记录的缓慢端点。最坏的情况是，如果两个用户大致同时呼叫端点，但他们不了解彼此的患者，您将遇到时间问题。

标签： python sqlalchemy fastapi python-typing

【解决方案1】：

在“患者.py”文件中

变化：

"patients = List[schemas.PatientInDb]"

到

"patients: List[schemas.PatientInDb]"

原因，“=”用于分配默认值，“:”用于引用类型。

【讨论】：

嘿，对不起。我纠正了那个错误。如果我遍历列表并通过将每个元素解析为 sqlalchemy 模型一次上传一个元素，它现在可以正常工作。有没有更好的方法来做到这一点？（也许通过避免 for 循环）

【解决方案2】：

您在 discord 上也得到了这个答案，但您保存的是 pydantic 模型而不是 sqlalchemy 模型。

【讨论】：

当表中的其他字段可以为空时，我们可以使用 pydantic 模型进行保存。无论如何，上面列出的错误的原因是因为参数中的不切实际的分配。