如何通过Python和Pandas DataFrame向SQLite插入唯一新数据？

问题 #

通过Python脚本和Pandas DataFrame向SQLite数据库插入外汇汇率数据，首次插入正常，但重复运行时要么生成重复行，要么覆盖原有数据。由于没有单一可作为主键的列，需要实现仅插入所有字段完全不重复的全新行，忽略已有重复数据。

现有代码流程：调用API获取JSON数据并转为DataFrame，连接SQLite创建表后逐行插入。尝试df.to_sql的if_exists='append'（重复插入）和if_exists='replace'（覆盖数据）参数均不满足需求。

核心代码片段：

# API获取数据并转为DataFrame
url = f'https://min-api.cryptocompare.com/data/{timeframe}?fsym={coin}&tsym={fx_converter}&limit={limiter}'
data = json.loads(requests.get(url).text)
df = pd.json_normalize(data, ['Data'])

# SQLite连接与插入逻辑
cnxn = sqlite3.connect("fx_rates.db")
cursor = cnxn.cursor()

# 创建表语句
table = f""" CREATE TABLE IF NOT EXISTS {coin} 
    (
        time                INTEGER NOT NULL,
        high                REAL,
        low                 REAL,
        open                REAL,
        volumefrom          INTEGER,
        volumeto            INTEGER,
        close               REAL,
        conversionType      TEXT,
        conversionSymbol    TEXT,
        date                TEXT
    )"""
cursor.execute(table)
cnxn.commit()

# 逐行插入
col = tuple(df.columns)
for i, value in df.iterrows():
    cursor.execute(
    f"""
        INSERT OR IGNORE INTO {coin}{col} 
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, 
    (
        value['time'],
        value['high'],
        value['low'],
        value['open'],
        value['volumefrom'],
        value['volumeto'],
        value['close'],
        value['conversionType'],
        value['conversionSymbol'],
        value['date']
    )
)
cnxn.commit()
cnxn.close()

可行解决方案 #

方法1：添加复合唯一约束 + INSERT OR IGNORE #

SQLite支持复合唯一约束，将所有字段组合作为唯一判断条件，配合INSERT OR IGNORE语法即可自动忽略完全重复的行，这是最高效的方案。

步骤：

1. 修改表创建语句，添加复合唯一约束
2. 使用df.to_sql（推荐，无需逐行循环）或原INSERT OR IGNORE逻辑插入数据

修改后的表创建代码：

table = f""" CREATE TABLE IF NOT EXISTS {coin} 
    (
        time                INTEGER NOT NULL,
        high                REAL,
        low                 REAL,
        open                REAL,
        volumefrom          INTEGER,
        volumeto            INTEGER,
        close               REAL,
        conversionType      TEXT,
        conversionSymbol    TEXT,
        date                TEXT,
        -- 添加复合唯一约束，所有字段完全匹配才判定为重复
        CONSTRAINT unique_full_row UNIQUE (time, high, low, open, volumefrom, volumeto, close, conversionType, conversionSymbol, date)
    )"""
cursor.execute(table)
cnxn.commit()

之后直接用df.to_sql插入，自动忽略重复行：

# 高效插入，无需逐行循环
df.to_sql(coin, cnxn, if_exists='append', index=False)

注意：若原表已存在，需先删除重建（SQLite的ALTER TABLE添加约束限制较多，重建更简便）。

方法2：临时表对比插入（无需修改原表） #

如果无法修改原表结构，可先将新数据导入临时表，再通过SQL对比原表，仅插入不存在的行。

代码示例：

cnxn = sqlite3.connect("fx_rates.db")
cursor = cnxn.cursor()

# 1. 将新数据导入临时表
temp_table = f"{coin}_temp"
df.to_sql(temp_table, cnxn, if_exists='replace', index=False)

# 2. 对比原表与临时表，仅插入未存在的行
insert_query = f"""
INSERT INTO {coin}
SELECT t.* FROM {temp_table} t
LEFT JOIN {coin} c 
ON t.time = c.time 
AND t.high = c.high 
AND t.low = c.low 
AND t.open = c.open 
AND t.volumefrom = c.volumefrom 
AND t.volumeto = c.volumeto 
AND t.close = c.close 
AND t.conversionType = c.conversionType 
AND t.conversionSymbol = c.conversionSymbol 
AND t.date = c.date
WHERE c.time IS NULL;
"""
cursor.execute(insert_query)
cnxn.commit()

# 3. 清理临时表
cursor.execute(f"DROP TABLE IF EXISTS {temp_table}")
cnxn.commit()
cnxn.close()

方法3：DataFrame前置过滤重复行 #

先从数据库读取已有数据，与新数据对比，仅保留全新行再插入，适合数据量较小的场景。

代码示例：

cnxn = sqlite3.connect("fx_rates.db")

# 1. 读取数据库中已有数据
existing_df = pd.read_sql(f"SELECT * FROM {coin}", cnxn)

# 2. 合并数据并过滤出仅存在于新数据中的行
merged = df.merge(existing_df, how='outer', indicator=True)
new_rows = merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)

# 3. 插入过滤后的新数据
new_rows.to_sql(coin, cnxn, if_exists='append', index=False)

cnxn.commit()
cnxn.close()

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