Skip to content

Instantly share code, notes, and snippets.

@kung-foo
Created February 3, 2012 15:13
Show Gist options
  • Save kung-foo/1730627 to your computer and use it in GitHub Desktop.
Save kung-foo/1730627 to your computer and use it in GitHub Desktop.
ON DUPLICATE KEY UPDATE hack
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)
@areshand
Copy link

The problem is that it will modify all the insert statements. Is there a way to avoid this ?

@amirbehzad
Copy link

amirbehzad commented Aug 22, 2017

@areshand try this:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert, 'postgresql')
def ignore_duplicates(insert, compiler, **kw):
    s = compiler.visit_insert(insert, **kw)
    ignore = insert.kwargs.get('postgresql_ignore_duplicates', False)
    return s if not ignore else s + ' ON CONFLICT DO NOTHING'
Insert.argument_for('postgresql', 'ignore_duplicates', None)

Sample usage:

stmt = MyModel.__table__.insert(postgresql_ignore_duplicates = True)

# -- OR --

from sqlalchemy import insert
stmt = insert(MyModel, postgresql_ignore_duplicates = True)

@631068264
Copy link

All the method above can't work.

@compiles(Insert, 'mysql')
def mysql_upsert(insert, compiler, **kw):
    def _gen_fv_dict(fv_dict):
        pass

    s = compiler.visit_insert(insert, **kw)
    if 'on_duplicate_key_update' in insert.kwargs:
        return s + ' ON DUPLICATE KEY UPDATE ' + _gen_fv_dict(insert.kwargs['on_duplicate_key_update'])
    return s

Insert.argument_for("mysql", "mysql_upsert", None)

 stmt = User.__table__.insert(on_duplicate_key_update={
        'age': 'age + VALUES(age)',
    })

session.execute(stmt, data)

And get error

SAWarning: Can't validate argument 'on_duplicate_key_update'; can't locate any SQLAlchemy dialect named 'on'
  % (k, dialect_name)

sqlalchemy.exc.IntegrityError: (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'bulko11' for key 'name'")
[SQL: INSERT INTO user (name, age) VALUES (%s, %s)]
[parameters: ('bulko11', 11)]

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment