Postgresql ON CONFLICT in sqlalchemy

Using Postgres 9.6.1, sqlachemy 1.1.4, and psycopg2 2.6.2:

1. Convert your data structure to a dictionary. From Pandas it is

   import pandas
   from sqlalchemy import MetaData
   from sqlalchemy.dialects.postgresql import insert
   import psycopg2
   
   # The dictionary should include all the values including index values
   insrt_vals = df.to_dict(orient='records')
   

2. Connect to database through sqlalchemy . Instead try psycog2 driver underneath and the native COPY function, which bypasses all the postgres indexing.

   csv_data = os.path.realpath('test.csv')
   con = psycopg2.connect(database = 'db01', user = 'postgres')
   cur = con.cursor()
   cur.execute("\copy stamm_data from '%s' DELIMITER ';' csv header" % csv_data)
   con.commit()
   

3. Execute

   results = engine.execute(do_nothing_stmt)
   # Get number of rows inserted
   rowcount = results.rowcount
   

Warning:

This method does not work with NaTs out of the box.

Everything together

tst_df = pd.DataFrame({'colA':['a','b','c','a','z', 'q'],
              'colB': pd.date_range(end=datetime.datetime.now() , periods=6),
              'colC' : ['a1','b2','c3','a4','z5', 'q6']})


insrt_vals = tst_df.to_dict(orient='records')
engine =      sqlalchemy.create_engine("postgresql://user:password@localhost/postgres")
connect = engine.connect()
meta = MetaData(bind=engine)
meta.reflect(bind=engine)
table = meta.tables['tstbl']
insrt_stmnt = insert(table).values(insrt_vals)

do_nothing_stmt  = insrt_stmnt.on_conflict_do_nothing(index_elements=['colA','colB'])
results = engine.execute(do_nothing_stmt)

Instead of step 2 and 3 , using psycog2 driver with the copy command in postgres is faster for larger files (approaching a gig) because it sets all the table indexing off.

csv_data = os.path.realpath('test.csv')

This works with Postgresql 10.5 and Sqlalchemy 1.3.6:

from sqlalchemy.dialects.postgresql import insert


table_info = {
'tableTime': '',
'deploymentID': '',
'tableData': ''
}
insert_table = insert(Table).values(table_info)
insert_table_sql = insert_table.on_conflict_do_nothing(
  index_elements=['tableTime', 'deploymentID']
)
db.session.execute(insert_table_sql)
db.session.commit()

This works with Postgresql 9.5:

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

@compiles(Insert)
def prefix_inserts(insert, compiler, **kw):
    return compiler.visit_insert(insert, **kw) + " ON CONFLICT DO NOTHING"

I use it for bulk_insert_mappings. It does however not make ON CONFLICT DO NOTHING optional

This is an extension of Niklas' answer.

Basically, uses a thread local state and a context manager to make appending the ON CONFLICT DO NOTHING optional. It's still a big hack though.

It also only hooks the postgres specific statement, and doesn't construct the sql query manually from text.

import threading
from contextlib import contextmanager

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert
from sqlalchemy.dialects.postgresql.dml import OnConflictDoNothing

state = threading.local()

@contextmanager
def on_conflict_do_nothing():
  state.active = True
  yield
  del state.active

@compiles(Insert, 'postgresql')
def prefix_inserts(insert, compiler, **kw):
  if getattr(state, "active", False):
    insert._post_values_clause = OnConflictDoNothing()
  return compiler.visit_insert(insert, **kw)