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

Question

我想公开一个API方法，该方法在POST请求中接收数据(用于测试版API注册(，并在已经有匹配模型时使用新值重新启动。实现这一目标的最佳方式是什么？目前我正在做这件事(有点简化(：

我的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发布方法：

@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((的一个问题是，它依赖于直接与电子邮件地址匹配的主键-我更愿意使用代理键，这样我就可以拥有禁用/删除功能，而不会被迫在数据库级别对电子邮件地址有唯一的约束。

Answer 1

基于FastAPI sqlalchemy演示应用程序，以下是解决此问题的方法：

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使用的命令式定义(

完整源代码

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

希望能有所帮助。(我花了太长时间搞清楚了。(

Answer 2

如果您使用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)

关于文档的更多信息

Answer 3

这是一个旧版本，但现在开始更新@mikey-no的答案。您不需要db.add()，因为您正在更新现有记录。您还可以循环使用Pydantic模型的字典表示，而不是使用vars(可能还包括"private"字段(。

所以，就像这样：

def update_user(db: Session, user: PydanticUserUpdate):
new_user = False
# get the existing data
db_user = db.query(User).filter(User.id == user.id).one_or_none()
# If user does not exist, create a new object
if db_user is None:
new_user = True
db_user = BetaORM()
# Update model class variable from requested fields 
for key, value in user.dict().items():
setattr(db_user, var, value) if value else None
if new_user:
db.add()
db.commit()
db.refresh(db_user)
return db_user

小调整，但很高兴知道。