Adding (Insert or update if key exists) option to `.to_sql`

[cdagnino]

Suppose you have an existing SQL table called person_age, where id is the primary key:

    age
id	
1	18
2	42

and you also have new data in a DataFrame called extra_data

	age
id	
2	44
3	95

then it would be useful to have an option on extra_data.to_sql() that allows to pass the DataFrame to SQL with an INSERT or UPDATE option on the rows, based on the primary key.

In this case, the id=2 row would get updated to age=44 and the id=3 row would get added

Expected Output

	age
id	
1	18
2	44
3	95

(Maybe) helpful code references

• Use merge from SQLAlchemy?
• The query: '''INSERT or REPLACE into person_age (id, age) values (?,?,?) ''' in this question

I looked at pandas sql.py sourcecode to come up with a solution, but I couldn't follow.

Code to replicate the example above

(Apologies for mixing sqlalchemy and sqlite

import pandas as pd
from sqlalchemy import create_engine
import sqlite3
conn = sqlite3.connect('example.db')
 
c = conn.cursor()
c.execute('''DROP TABLE IF EXISTS person_age;''')
c.execute('''
          CREATE TABLE person_age
          (id INTEGER PRIMARY KEY ASC, age INTEGER NOT NULL)
          ''')
conn.commit()
conn.close()

##### Create original table

engine = create_engine("sqlite:///example.db")
sql_df = pd.DataFrame({'id' : [1, 2], 'age' : [18, 42]})

sql_df.to_sql('person_age', engine, if_exists='append', index=False)


#### Extra data to insert/update

extra_data = pd.DataFrame({'id' : [2, 3], 'age' : [44, 95]})
extra_data.set_index('id', inplace=True)

#### extra_data.to_sql()  with row update or insert option

expected_df = pd.DataFrame({'id': [1, 2, 3], 'age': [18, 44, 95]})
expected_df.set_index('id', inplace=True)

[jorisvandenbossche]

This would be nice functionality, but the main problem is that we want it to be database-flavor independent and based on sqlalchemy core (so not sqlalchemy ORM) for inclusion in pandas itself.
Which will make this difficult to implement ..

[TomAugspurger]

Yeah, I think this is out of scope for pandas since upserts aren't supported by all db engines.

[kjford]

While an INSERT OR UPDATE isn't supported by all engines, an INSERT OR REPLACE can be made engine agnostic by deleting rows from the target table for the set of primary keys in the DataFrame index followed by an insert of all rows in the DataFrame. You'd want to do this in a transaction.

[neilfrndes]

@TomAugspurger Could we add the upsert option for supported db engines and throw an error for unsupported db engines ?

[ldacey]

I'd like to see this as well. I am caught in between using pure SQL and SQL Alchemy (haven't gotten this to work yet, I think it has something to do with how I pass the dicts). I use psycopg2 COPY to bulk insert, but I would love to use pd.to_sql for tables where values might change over time and I don't mind it inserting a bit slower.

insert_values = df.to_dict(orient='records')
insert_statement = sqlalchemy.dialects.postgresql.insert(table).values(insert_values)
upsert_statement = insert_statement.on_conflict_do_update(
    constraint='fact_case_pkey',
    set_= df.to_dict(orient='dict')
)

And pure SQL:

def create_update_query(df, table=FACT_TABLE):
    """This function takes the Airflow execution date passes it to other functions"""
    columns = ', '.join([f'{col}' for col in DATABASE_COLUMNS])
    constraint = ', '.join([f'{col}' for col in PRIMARY_KEY])
    placeholder = ', '.join([f'%({col})s' for col in DATABASE_COLUMNS])
    values = placeholder
    updates = ', '.join([f'{col} = EXCLUDED.{col}' for col in DATABASE_COLUMNS])
    query = f"""INSERT INTO {table} ({columns}) 
    VALUES ({placeholder}) 
    ON CONFLICT ({constraint}) 
    DO UPDATE SET {updates};"""
    query.split()
    query = ' '.join(query.split())
    return query

def load_updates(df, connection=DATABASE):
    """Uses COPY from STDIN to load to Postgres
     :param df: The dataframe which is writing to StringIO, then loaded to the the database
     :param connection: Refers to a PostgresHook
    """
    conn = connection.get_conn()
    cursor = conn.cursor()
    df1 = df.where((pd.notnull(df)), None)
    insert_values = df1.to_dict(orient='records')
    for row in insert_values:
        cursor.execute(create_update_query(df), row)
        conn.commit()
    cursor.close()
    del cursor
    conn.close()

[ODemidenko]

@ldacey this style worked for me (insert_statement.excluded is an alias to the row of data that violated the constraint):

insert_values = merged_transactions_channels.to_dict(orient='records')
 insert_statement = sqlalchemy.dialects.postgresql.insert(orders_to_channels).values(insert_values)
    upsert_statement = insert_statement.on_conflict_do_update(
        constraint='orders_to_channels_pkey',
        set_={'channel_owner': insert_statement.excluded.channel_owner}
    )

[rajbiswas]

@cdagnino This snippet might not work in the case of composite keys, that scenario has to be taken care of also. I'll try to find a way to do the same

[danich1]

One way to solve this update issue is to use sqlachemy's bulk_update_mappings. This function takes in a list of dictionary values and updates each row based on the tables primary key.

session.bulk_update_mappings(
  Table,
  pandas_df.to_dict(orient='records)
)

[joshhornby]

I agree with @neilfrndes, shouldn't allow a nice feature like this not to be implemented because some DBs don't support. Is there any chance this feature might happen?

[TomAugspurger]

Probably. if someone makes a PR. On further consideration, I don't think I'm opposed to this on the principle that some databases don't support it. However, I'm not too familiar with the sql code, so I'm not sure what the best approach is.

[kjford]

One possibility is to provide some examples for upserts using the method callable if this PR is introduced: #21401

For postgres that would look something like (untested):

from sqlalchemy.dialects import postgresql

def pg_upsert(table, conn, keys, data_iter):
    for row in data:
        row_dict = dict(zip(keys, row))
        stmt = postgresql.insert(table).values(**row_dict)
        upsert_stmt = stmt.on_conflict_do_update(
            index_elements=table.index,
            set_=row_dict)
        conn.execute(upsert_stmt)

Something similar could be done for mysql.