MySQL on duplicate key in SQLAlchemy
Posted | stdout
A simple function to upsert something atomically into MySQL
def upsert(val1, val2, **kwargs):
"""upsert val1-val2 pair into mysql and return the pk"""
from sqlalchemy.dialects.mysql import insert
from sqlalchemy.sql.expression import func
# this val1-val2 pair should have UNIQUE KEY constraint in MySQL table.
stmt = insert(MyModel.__table__).values(col1=val1, col2=val2)
if kwargs:
stmt = stmt.on_duplicate_key_update(kwargs)
else:
# get lastrowid https://stackoverflow.com/a/29722203/41948
stmt = stmt.on_duplicate_key_update(id=func.LAST_INSERT_ID(MyModel.id))
# inserted_primary_key (dbapi, return 0 when only insert) or lastrowid (mysql)
pk = db.session().execute(stmt).lastrowid
if not pk:
pk = MyModel.query.options(load_only('id')).filter_by(
col1=val1, col2=val2
).first().id
return pk
Writing complex SQL is hard, sqlalchemy makes it harder. Prefer other ORMs in Python.
I really hate fighting the infamous sqlalchemy API. Everything wrong about it can be shown in this one-liner:
from sqlalchemy.orm import Load, load_only, joinedload
You have the three styles of CamelCase, under_score, and concatfunctionnames fucked up together. Bad and leaky abstraction.
Comments