时间戳

首页 > 解决方案 > 从 FastAPI 中发布的 Pydantic 模型更新 SQLAlchemy ORM 现有模型？

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

问题描述

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

标签： pythonsqlalchemyfastapipydantic

解决方案

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

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

推荐阅读