Update SQLAlchemy ORM existing model from posted Pydantic model in FastAPI?

Question

I want to expose an API method that receives data in a POST request (for a beta signup API), and upsert with new values when there's already a matching model. What's the best way to achieve this? Currently I'm doing this (somewhat simplified):

My ORM model (SqlAlchemy):

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

My Pydantic model:

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

My FastAPI post method:

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

And the CRUD method I've written:

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

One problem with using merge() is that it relies on matching with primary key, directly against the email address - I'd much rather use a surrogate key instead, so that I can have a disable / delete functionality and not be forced to have a unique constraint on the email address at the database level.

Answer 1

Based on the FastAPI sqlalchemy demo application, this is how a solved this problem:

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

This approach uses the SQLAlchemy declarative model definition (not the imperative definition as Gabriel Cappelli uses)

Full source code

I used this approach to do an update in crud in a FastAPI based application. Existing data is left unchanged and new update values overwrite. Modified datetime is updated (but for ease of testing this value is fixed).

Hope it helps. (I spent too long figuring this out.)

Answer 2

If you're using MySQL and SQLAlchemy >= 1.2 you can use INSERT...ON DUPLICATE KEY UPDATE using SQLAlchemy.

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)

More info on the docs