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 builds a single INSERT ... ON CONFLICT ... DO UPDATE statement. Values never enter the SQL string: each row supplies one %s placeholder per column, and execute_batch passes the values to psycopg2 separately, which prevents SQL injection through the data ¹. The table and column names cannot be passed as parameters, so they are composed with the psycopg2.sql module, which quotes each one as an identifier rather than pasting it into the query.

Two details matter. ON CONFLICT names the columns of a unique constraint, and EXCLUDED refers to the row that failed to insert, so col = EXCLUDED.col updates the existing row without sending every value twice. And psycopg2 cannot adapt NumPy types such as numpy.int64, so the DataFrame is converted to plain Python objects, with missing values mapped to None (NULL), before it is sent.

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

def upsert_dataframe(conn, df, table, conflict_cols, 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
    conflict_cols: list of str
        Columns of the unique constraint used to determine insert or update
    page_size: int
        Number of insert statements per command.

    Returns
    -------
    None

    """
    # Convert numpy types to Python types and NaN to None (NULL)
    tuples = list(df.astype(object).where(df.notna(), None).itertuples(index=False, name=None))
    cols = list(df.columns)
    update_cols = [col for col in cols if col not in conflict_cols]
    # SQL query to execute, with identifiers quoted by psycopg2
    query = sql.SQL("""
    INSERT INTO {table} ({cols}) VALUES ({placeholders})
    ON CONFLICT ({conflict_cols})
    DO UPDATE SET {updates}
    """).format(
        table=sql.Identifier(table),
        cols=sql.SQL(",").join(map(sql.Identifier, cols)),
        placeholders=sql.SQL(",").join(sql.Placeholder() * len(cols)),
        conflict_cols=sql.SQL(",").join(map(sql.Identifier, conflict_cols)),
        updates=sql.SQL(",").join(
            sql.SQL("{0} = EXCLUDED.{0}").format(sql.Identifier(col)) for col in update_cols
        ),
    )
    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} on columns {1} complete".format(table, conflict_cols))
    cursor.close()

Quoting identifiers stops a malicious table or column name from rewriting the statement, but it does not check that the name is one you expect: the caller still decides which table is written to. Keep table and column names under your own control rather than taking them from user input.

Footnotes

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