CODEWITHSUNDEEP
pythonsqlalchemyfastapipydantic
Dhiraj Gupta
Dhiraj Gupta

Reputation: 10504

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

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.

Upvotes: 15

Views: 23634

Answers (3)

mikey-no
mikey-no

Reputation: 319

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 pydantic 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 are overwriten. Modified datetime is updated (but for ease of testing this value is fixed).

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

Upvotes: 15

Erik van de Ven
Erik van de Ven

Reputation: 4985

It is an old one but here you go, an update on @mikey-no's answer. You don't need db.add() as you are updating an existing record. You can also just loop over the dictionary representation of the Pydantic model, instead of using vars (which also might include "private" fields).

So, just like this:

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

Small adjustments, but good to know.

Upvotes: 4

Gabriel Cappelli
Gabriel Cappelli

Reputation: 4190

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

Upvotes: 2

Related Questions