How to use SQLite's UPSERT (or ON CONFLICT) with flask_sqlalchemy?

user_8376337 0
Viewed 1622

I want to update my SQLite DB with data provided by an external API and I don't want to check every time for any conflicts by myself, instead, I want to take advantage of UPSERT statement. SQLAlchemy documentation for version 1.4 (still in beta but that's ok) shows that this is possible. But I don't know how to get this to work with flask_sqlalchemy.

I've also tried to include sqlite_on_conflict='IGNORE' in constraint definitions (as mentioned in SQLAlchemy docs for version 1.3 here):

class SomeModel(db.Model):
    __table_args__ = (
        db.ForeignKeyConstraint(
            ('id', 'other_id'),
            ('other_table.id', 'other_table.other_id')
            sqlite_on_conflict='IGNORE'
        ),
    )
    # ...

I then verified SQL output with SQLALCHEMY_ECHO set to True and it didn't work at all...

I tried with both 1.3 and 1.4 SQLAlchemy versions.

flask-sqlalchemy sqlalchemy sqlite python
1 Answers

I found the solution thanks to this gist (credits for droustchev) which I slightly edited. It's a bit messy but it works:

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


@compiles(Insert, 'sqlite')
def suffix_insert(insert, compiler, **kwargs):
    stmt = compiler.visit_insert(insert, **kwargs)
    if insert.dialect_kwargs.get('sqlite_on_conflict_do_nothing'):
        stmt += ' ON CONFLICT DO NOTHING'
    return stmt


Insert.argument_for('sqlite', 'on_conflict_do_nothing', False)

Just include this code in your models.py file (or whatever filename you have) and then you can write:

some_random_values = [
  {'id': 1, 'some_column': 'asadsadad'},
  {'id': 2, 'some_column': 'dsawaefds'}
]

stmt = db.insert(MyModel, sqlite_on_conflict_do_nothing=True).values(some_random_values)
db.session.execute(stmt)

This should do the trick.

user_83763370
Related