Skip to content

Instantly share code, notes, and snippets.

@droustchev
Forked from kung-foo/gist:1730627
Last active August 6, 2022 23:34
Show Gist options
  • Save droustchev/5145cb1fcbf0cf365229 to your computer and use it in GitHub Desktop.
Save droustchev/5145cb1fcbf0cf365229 to your computer and use it in GitHub Desktop.
SQLAlchemy extension for "ON DUPLICATE KEY UPDATE"
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert
@compiles(Insert)
def append_string(insert, compiler, **kw):
s = compiler.visit_insert(insert, **kw)
if 'append_string' in insert.kwargs:
return s + " " + insert.kwargs['append_string']
return s
my_connection.execute(my_table.insert(append_string = 'ON DUPLICATE KEY UPDATE foo=foo'), my_values)
# for newer version 0.9+
# see: https://bitbucket.org/zzzeek/sqlalchemy/issue/3113/warning-for-custom-insert-mysql
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import Insert
@compiles(Insert, 'mysql')
def suffix_insert(insert, compiler, **kw):
stmt = compiler.visit_insert(insert, **kw)
if 'mysql_on_duplicate_key_update_cols' in insert.dialect_kwargs:
my_var = insert.kwargs['mysql_on_duplicate_key_update_cols']
stmt += ' ON DUPLICATE KEY UPDATE %s' % (", ".join(my_var))
return stmt
Insert.argument_for("mysql", "on_duplicate_key_update_cols", None)
if __name__ == '__main__':
from sqlalchemy.sql import table, column
from sqlalchemy.dialects import mysql
print table('t', column('x')).insert(
mysql_on_duplicate_key_update_cols=['x']).compile(
dialect=mysql.dialect())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment