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.
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)
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.)
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