MySQL on duplicate keys in SQLAlchemy

Posted 2021-06-29 | 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)
    r = db.session().execute(stmt).lastrowid
    if not r:
        r = MyModel.query.options(load_only('id')).filter_by(
            col1=val1, col2=val2
        ).first().id
    return r

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.