8

我想公开一个在 POST 请求中接收数据的 API 方法（用于 beta 注册 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() 的一个问题是它依赖于与主键匹配，直接针对电子邮件地址 - 我宁愿使用代理键，这样我就可以拥有禁用/删除功能而不是被迫拥有数据库级别的电子邮件地址的唯一约束。

4

2 回答 2

8

基于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 更新。现有数据保持不变，并覆盖新的更新值。修改后的日期时间已更新（但为了便于测试，此值是固定的）。

希望能帮助到你。（我花了太长时间才弄清楚这一点。）

于 2021-02-02T21:08:24.847 回答
1

如果您使用 MySQL 和 SQLAlchemy >= 1.2，您可以使用INSERT...ON DUPLICATE KEY UPDATESQLAlchemy。

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)

有关文档的更多信息

于 2020-08-05T03:13:15.080 回答