How to Safely Upsert DataFrames into Postgres

Moving data between applications and Postgres is a common task. I found this blog post particularly useful—it compared insertion methods using psycopg2 and pandas DataFrames. I adapted one approach to create a robust upsert function that respects database constraints.

Upserts in Postgres update existing rows and insert new ones—a single operation that avoids the complexity of checking which rows exist first. When two rows conflict on a database constraint, use the ON CONFLICT clause to specify the desired behavior. Since Pandas lacks native upsert support, you must construct these queries manually.

The function below constructs two SQL strings: one for the INSERT statement and one for the UPDATE clause (executed on conflict). The code uses string interpolation to build the template ¹, but passes actual values separately to execute_batch to prevent SQL injection.

import logging
import psycopg2
import psycopg2.extras
import pandas as pd

def upsert_dataframe(conn, df, table, constraint_name, page_size=100):
    """
    Upserting a pandas dataframe into a Postgres database.

    Parameters
    ----------
    conn: psycopg2.connection
        Connection to postgres database.
    df: pd.DataFrame
        Pandas dataframe for upsert
    table: str
        Name of table that dataframe is to be upserted into
    constraint_name: str
        Name of the constraint used to determine insert or update
    page_size: int
        Number of insert statements per command.

    Returns
    -------
    None

    """
    # Create a list of tuples from the dataframe values
    tuples = [tuple(x)*2 for x in df.to_numpy()]
    # Comma-separated dataframe columns
    cols = ','.join(list(df.columns))
    # SQL query to execute
    insert_stmt  = """
    INSERT INTO {0} ({1}) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
    ON CONFLICT ON CONSTRAINT {2}
    DO UPDATE SET
    """.format(table, cols, constraint_name)
    update_stmt = ",".join(["{0}=%s".format(col) for col in list(df.columns)])
    query = insert_stmt + update_stmt
    cursor = conn.cursor()
    try:
        psycopg2.extras.execute_batch(cursor, query, tuples, page_size)
        conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        print("Error: %s" % error)
        conn.rollback()
        cursor.close()
        return 1
    logging.info("Upsert to table {0} using constraint {1} complete".format(table, constraint_name))
    cursor.close()

Footnotes

1. String interpolation creates SQL injection vulnerabilities. Always separate query templates from values. The Psycopg2 documentation explains this problem in detail. ↩