将DataFrame保存至带约束的SQLite表时遇重复键问题求助

当assignments表以employee_id+project_id作为复合主键时，直接使用to_sql(if_exists='append')会因主键重复抛出约束错误。要实现“丢弃重复行、保存其余行”的需求，有两种实用方案：

方案1：先过滤DataFrame中的重复行（适合小数据量） #

先从数据库读取已存在的主键组合，在DataFrame中过滤掉这些重复项后再插入。

import pandas as pd
import sqlalchemy

engine = sqlalchemy.create_engine('sqlite:///my_database.db')

# 1. 查询数据库中已有的复合主键组合
with engine.connect() as conn:
    existing_keys = conn.execute(sqlalchemy.text(
        "SELECT employee_id, project_id FROM assignments"
    )).fetchall()
# 转成集合提升查找效率
existing_set = set(existing_keys)

# 2. 构造待插入的DataFrame
data = {
    'employee_id': [1, 2],
    'project_id': ['ABC123', 'DEF456']
}
df_new = pd.DataFrame(data)

# 3. 过滤掉已存在的行
df_to_insert = df_new[~df_new.apply(lambda row: (row['employee_id'], row['project_id']) in existing_set, axis=1)]

# 4. 插入数据库（空DataFrame时跳过）
if not df_to_insert.empty:
    df_to_insert.to_sql('assignments', engine, if_exists='append', index=False)

方案2：使用SQLite的INSERT OR IGNORE语法（适合大数据量） #

利用SQLite原生的INSERT OR IGNORE语句，直接忽略主键冲突的行，无需提前查询数据库，性能更优。通过自定义method参数实现：

import pandas as pd
import sqlalchemy
from sqlalchemy import text

engine = sqlalchemy.create_engine('sqlite:///my_database.db')

data = {
    'employee_id': [1, 2],
    'project_id': ['ABC123', 'DEF456']
}
df_new = pd.DataFrame(data)

# 自定义插入方法，使用INSERT OR IGNORE跳过冲突行
def insert_or_ignore(table, conn, keys, data_iter):
    # 构造INSERT OR IGNORE语句
    insert_stmt = text(
        f"INSERT OR IGNORE INTO {table.name} ({', '.join(keys)}) VALUES ({', '.join([':'+k for k in keys])})"
    )
    # 批量执行插入
    conn.execute(insert_stmt, [dict(zip(keys, row)) for row in data_iter])

# 调用to_sql时指定自定义方法
df_new.to_sql(
    'assignments', 
    engine, 
    if_exists='append', 
    index=False,
    method=insert_or_ignore
)

说明： #

• INSERT OR IGNORE会自动跳过违反主键/唯一约束的行，既不抛出错误也不修改已有数据
• 方案2无需提前查询数据库，在数据量较大时性能优势明显

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