从FastAPI中发布的Pydantic模型更新SQLAlchemy ORM现有模型？

Question

我想公开一个API方法，该方法在POST请求中接收数据（对于beta注册API），并在已经存在匹配模型时使用新值进行upsert。实现这一目标的最佳方式是什么？目前我正在这样做（有点简化）：

我的ORM模型（SqlAlchemy）：

class BetaORM(Base):
    __tablename__ = "betasignup"
    email = Column(EmailType, primary_key=True)
    fullname = Column(String, unique=False, index=False, nullable=True)

我的Pydantic模型：

class BetaCreate(BaseModel):
    email: EmailStr
    fullname: Optional[str]

我的FastAPI post方法：

@app.post("/beta_signup")
def post_beta_signup(beta: schemas.BetaCreate, db: Session = Depends(get_db)):
    return create_beta_signup(db=db,signup=beta)

还有我写的CRUD方法：

def create_beta_signup(db: Session, signup: schemas.BetaCreate):
    db_beta = schemas.BetaORM(**signup.dict())
    ret_obj = db.merge(db_beta)
    db.add(ret_obj)
    db.commit()
    return ret_obj

使用merge（）的一个问题是，它依赖于直接针对电子邮件地址与主键进行匹配——我更愿意使用surrogate key，这样我就可以拥有禁用/删除功能，而不必在数据库级别强制对电子邮件地址进行唯一约束

Answer 1

如果您使用的是MySQL和SQLAlchemy>；=1.2您可以使用SQLAlchemy使用INSERT...ON DUPLICATE KEY UPDATE

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='inserted value')

on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
    data=insert_stmt.inserted.data,
    status='U'
)

conn.execute(on_duplicate_key_stmt)

有关docs的更多信息

Answer 2

基于FastAPI sqlalchemy demo application，以下是a解决此问题的方法：

def update_user(db: Session, user: PydanticUserUpdate):
    """
    Using a new update method seen in FastAPI https://github.com/tiangolo/fastapi/pull/2665
    Simple, does not need each attribute to be updated individually
    Uses python in built functionality... preferred to the pydintic related method
    """

    # get the existing data
    db_user = db.query(User).filter(User.id == user.id).one_or_none()
    if db_user is None:
        return None

    # Update model class variable from requested fields 
    for var, value in vars(user).items():
        setattr(db_user, var, value) if value else None

    db_user.modified = modified_now
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

这种方法使用SQLAlchemy声明性模型定义（而不是Gabriel Cappelli使用的命令式定义）

Full source code

我使用这种方法在基于FastAPI的应用程序中对crud进行更新。现有数据保持不变，新的更新值将覆盖。修改后的日期时间会更新（但为了便于测试，此值是固定的）

希望能有帮助。（我花了太长时间才弄明白这一点。）