cardinal_pythonlib.sqlalchemy.insert_on_duplicate

---

Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).

This file is part of cardinal_pythonlib.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Add “INSERT ON DUPLICATE KEY UPDATE” functionality to SQLAlchemy for MySQL.

OLD VERSION (before SQLAlchemy 1.4/future=True or SQLAlchemy 2.0):

• https://www.reddit.com/r/Python/comments/p5grh/sqlalchemy_whats_the_idiomatic_way_of_writing/

• https://github.com/bedwards/sqlalchemy_mysql_ext/blob/master/duplicate.py … modified

• https://docs.sqlalchemy.org/en/rel_1_0/core/compiler.html

• https://stackoverflow.com/questions/6611563/sqlalchemy-on-duplicate-key-update

• https://dev.mysql.com/doc/refman/5.7/en/insert-on-duplicate.html

Once implemented, you can do

q = sqla_table.insert_on_duplicate().values(destvalues)
session.execute(q)

Then: this partly superseded by SQLAlchemy v1.2:

• https://docs.sqlalchemy.org/en/latest/changelog/migration_12.html

• https://docs.sqlalchemy.org/en/latest/dialects/mysql.html#mysql-insert-on-duplicate-key-update

FOR SQLAlchemy 1.4/future=True OR SQLAlchemy 2.0:

New function: insert_with_upsert_if_supported().

cardinal_pythonlib.sqlalchemy.insert_on_duplicate.insert_with_upsert_if_supported(table: Table, values: Dict, session: Session | None = None, dialect: Dialect | None = None) → Insert

Creates an “upsert” (INSERT … ON DUPLICATE KEY UPDATE) statment if possible (e.g. MySQL/MariaDB). Failing that, returns an INSERT statement.

Parameters:

• table – SQLAlchemy Table in which to insert values.

• values – Values to insert (column: value dictionary).

• session – Session from which to extract a dialect.

• dialect – Explicit dialect.

Previously (prior to 2025-01-05 and prior to SQLAlchemy 2), we did this:

q = sqla_table.insert_on_duplicate().values(destvalues)

This “insert_on_duplicate” member was available because crate_anon/anonymise/config.py ran monkeypatch_TableClause(), from cardinal_pythonlib.sqlalchemy.insert_on_duplicate. The function did dialect detection via “@compiles(InsertOnDuplicate, SqlaDialectName.MYSQL)”. But it did nasty text-based hacking to get the column names.

However, SQLAlchemy now supports “upsert” for MySQL: https://docs.sqlalchemy.org/en/20/dialects/mysql.html#insert-on-duplicate-key-update-upsert

Note the varying argument forms possible.

The only other question: if the dialect is not MySQL, will the reference to insert_stmt.on_duplicate_key_update crash or just not do anything? To test:

from sqlalchemy import table
t = table("tablename")
destvalues = {"a": 1}

insert_stmt = t.insert().values(destvalues)
on_dup_key_stmt = insert_stmt.on_duplicate_key_update(destvalues)

This does indeed crash (AttributeError: ‘Insert’ object has no attribute ‘on_duplicate_key_update’). In contrast, this works:

from sqlalchemy.dialects.mysql import insert as insert_mysql

insert2 = insert_mysql(t).values(destvalues)
on_dup_key2 = insert2.on_duplicate_key_update(destvalues)

Note also that an insert() statement doesn’t gain a “on_duplicate_key_update” attribute just because MySQL is used (the insert statement doesn’t know that yet).

The old way was good for dialect detection but ugly for textual analysis of the query. The new way is more elegant in the query, but less for dialect detection. Overall, new way likely preferable.