如何通过Python的pandas.to_sql实现PostgreSQL行更新？

Great question! You're right that pandas' to_sql() doesn't have a built-in upsert (insert-or-update) feature, which is super frustrating when you need to keep your database tables in sync without duplicate entries. Your current approach of appending first then cleaning up duplicates works, but there are cleaner ways to handle this directly within your Python code. Let's break down the most practical solutions:

1. Use SQLAlchemy Core to Build a Native Upsert (Recommended) #

Since to_sql() relies on SQLAlchemy under the hood, you can leverage SQLAlchemy Core to construct a database-native upsert statement. This is the most efficient approach because it handles the logic in a single atomic database operation, no intermediate duplicates required.

Here's a PostgreSQL-focused example (adjust syntax for your database—e.g., ON DUPLICATE KEY UPDATE for MySQL, MERGE for SQL Server):

from sqlalchemy import create_engine, MetaData, Table
import pandas as pd

# Initialize your database connection
engine = create_engine("postgresql://user:password@host/db_name")
metadata = MetaData()

# Load the existing table schema from the database
dim_objects_table = Table("Dim_Objects", metadata, autoload_with=engine)

# Your target DataFrame with new/updated records
data = pd.DataFrame(...)

# Build the upsert statement
insert_stmt = dim_objects_table.insert()
upsert_stmt = insert_stmt.on_conflict_do_update(
    # Define the column(s) that trigger a conflict (your unique identifier)
    index_elements=["Code"],
    # Specify which columns to update when a conflict occurs
    set_={
        "Column1": insert_stmt.excluded.Column1,
        "Column2": insert_stmt.excluded.Column2,
        "TimeStampUpdate": insert_stmt.excluded.TimeStampUpdate
        # Add all other columns you want to update here
    }
)

# Execute the upsert
with engine.connect() as conn:
    conn.execute(upsert_stmt, data.to_dict("records"))
    conn.commit()

Pros: Fast, atomic, no intermediate duplicate data.
Cons: Syntax varies slightly by database, but SQLAlchemy abstracts most of the complexity.

2. Split Data into Insert/Update Batches #

If you prefer a more straightforward (though less efficient) approach, you can split your DataFrame into two groups: records that don't exist in the database (to insert) and records that do exist (to update).

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql://user:password@host/db_name")

# Your target DataFrame
data = pd.DataFrame(...)

# Fetch all existing Code values from the database
existing_codes = pd.read_sql("SELECT Code FROM Dim_Objects", engine)["Code"].tolist()

# Split data into new records (to insert) and existing records (to update)
new_records = data[~data["Code"].isin(existing_codes)]
update_records = data[data["Code"].isin(existing_codes)]

# Insert new records
if not new_records.empty:
    new_records.to_sql("Dim_Objects", con=engine, if_exists="append", index=False)

# Update existing records
if not update_records.empty:
    with engine.connect() as conn:
        for _, row in update_records.iterrows():
            update_query = """
                UPDATE Dim_Objects
                SET Column1 = %s, Column2 = %s, TimeStampUpdate = %s
                WHERE Code = %s
            """
            conn.execute(update_query, (row["Column1"], row["Column2"], row["TimeStampUpdate"], row["Code"]))
        conn.commit()

Pros: Logic is easy to follow, no deep SQLAlchemy knowledge needed.
Cons: Slow for large datasets (due to row-by-row updates), requires two separate database interactions.

3. Customize SQLTable for Reusable Upsert Logic #

For a more reusable solution, you can subclass pandas' SQLTable to override the insert method with upsert logic. This lets you create a drop-in replacement for to_sql() that handles upserts automatically.

from pandas.io.sql import SQLTable
from sqlalchemy.dialects.postgresql import insert

class UpsertSQLTable(SQLTable):
    def insert(self, chunksize=None, method=None):
        # Override the default insert method with upsert logic
        data = self.data
        table = self.table.tometadata(self.pd_sql.meta, schema=self.schema) if self.schema else self.table

        # Build the upsert statement
        insert_stmt = insert(table).values(data.to_dict("records"))
        upsert_stmt = insert_stmt.on_conflict_do_update(
            index_elements=["Code"],
            set_={col: insert_stmt.excluded[col] for col in data.columns if col != "Code"}
        )

        # Execute the upsert
        with self.pd_sql.connectable.begin() as conn:
            conn.execute(upsert_stmt)

# Create a wrapper function to use our custom table
def to_sql_upsert(df, name, con, **kwargs):
    table = UpsertSQLTable(
        name,
        con,
        frame=df,
        index=kwargs.get("index", True),
        schema=kwargs.get("schema", None),
        if_exists=kwargs.get("if_exists", "fail"),
        keys=kwargs.get("keys", None),
    )
    table.insert(chunksize=kwargs.get("chunksize", None))

# Usage example
to_sql_upsert(data, "Dim_Objects", con=engine, index=False)

Pros: Reusable, acts like the native to_sql() function.
Cons: Requires familiarity with pandas' internal SQL handling.

Overall, the first approach (using SQLAlchemy Core's native upsert syntax) is the best choice for most cases—it's efficient, atomic, and avoids the intermediate duplicate data you get with the append-then-delete method.

内容的提问来源于stack exchange，提问作者Hryhorii Pavlenko