如何用Pandas df.to_sql更新Django数据库重复行并插入新数据？

每日上传包含过去7天数据的文件，用Pandas读取生成DataFrame后同步至Django数据库。DataFrame中存在数据库已有的重复行（当date、advertiser、insertion_order三者与数据库中数据相同时，判定为重复行）和新行，需实现：用DataFrame中的行替换数据库重复行，同时插入新行。

现有代码 #

自定义upsert函数 #

def upsert(table, con, keys, data_iter):
    
    data = [dict(zip(keys, row)) for row in data_iter]
    insert_statement = insert(table.table).values(data)
    upsert_statement = insert_statement.on_conflict_do_update(
        constraint=f"{table.table.name}_pkey",
        set_={c.key: c for c in insert_statement.excluded},
    )

print(upsert_statement)
con.execute(upsert_statement)

df.to_sql调用代码 #

df.to_sql(
  Xandr._meta.db_table,
  if_exists="append",
  index=False,
  dtype=dtype,
  chunksize=1000,
  con=engine,
  method=upsert
)

Django模型代码 #

class Xandr(InsertionOrdersCommonFields):
    dsp = models.CharField("Xandr", max_length=20)

    class Meta:
        db_table = "Xandr"
        unique_together = [
            "date",
            "advertiser",
            "insertion_order"
        ]
        verbose_name_plural = "Xandr"

    def __str__(self):
        return self.dsp

问题分析 #

你的upsert函数无法工作的核心原因有两个：

1. 约束名错误：你指定的constraint=f"{table.table.name}_pkey"是主键约束，但你的Django模型并未将date、advertiser、insertion_order设为主键，而是用unique_together定义了唯一约束，数据库中对应的约束名不是主键名，导致on_conflict_do_update找不到冲突触发条件。
2. 代码缩进错误：print(upsert_statement)和con.execute(upsert_statement)不在函数体内，无法被执行。

解决方案 #

1. 修正upsert函数 #

根据你使用的数据库类型，选择对应的修正版本：

适配PostgreSQL

直接指定冲突的列（无需依赖约束名）：

from sqlalchemy import insert

def upsert(table, con, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]
    insert_stmt = insert(table.table).values(data)
    # 指定unique_together对应的三个列作为冲突判定依据
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["date", "advertiser", "insertion_order"],
        set_={col: insert_stmt.excluded[col] for col in keys}
    )
    con.execute(upsert_stmt)

适配MySQL

MySQL使用on_duplicate_key_update，通过unique_by指定冲突列：

from sqlalchemy import insert

def upsert(table, con, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]
    insert_stmt = insert(table.table).values(data)
    upsert_stmt = insert_stmt.on_duplicate_key_update(
        unique_by=["date", "advertiser", "insertion_order"],
        **{col: insert_stmt.excluded[col] for col in keys}
    )
    con.execute(upsert_stmt)

2. 验证数据库唯一约束 #

确保Django的unique_together已经同步到数据库：

• 如果你是通过Django迁移创建的表，唯一约束会自动生成；
• 若手动创建表，需手动添加唯一约束：
  • PostgreSQL：ALTER TABLE Xandr ADD CONSTRAINT xandr_unique_date_advertiser_io UNIQUE (date, advertiser, insertion_order);
  • MySQL：ALTER TABLE Xandr ADD UNIQUE KEY (date, advertiser, insertion_order);

3. 保持df.to_sql调用不变 #

你的df.to_sql调用参数无需修改，自定义的method会处理upsert逻辑。

额外注意事项 #

• 确保DataFrame的列名与数据库表的列名完全一致，否则会导致字段匹配失败；
• chunksize=1000可以根据数据量调整，避免单次插入数据过多导致性能问题；
• 如果需要查看生成的SQL语句，可以在con.execute(upsert_stmt)前添加print(upsert_stmt)（放在函数体内）。

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