使用cursor.executemany()插入数据时，如何定位引发IntegrityError异常的行

你遇到的这个情况太常见了！SQLite的executemany()在批量插入时，一旦碰到约束违反的异常就会直接终止整个批次，而且根本不会告诉你具体是哪一行出的问题——这确实挺闹心的。

不用急，下面给你几个实用的解决方案，你可以根据自己的数据量和场景来选：

方案一：逐行插入（简单直接，适合小数据量） #

如果你的数据量不大，最省心的办法就是放弃批量插入，改成逐行循环插入，每一行单独捕获异常。这样哪一行出错，你能立刻定位到。

修改你的代码试试：

import sqlite3

conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS test(name TEXT,UNIQUE(name))')

data = [('Larry',),('Curly',),('Moe',),('Larry',)]

success_count = 0
for idx, row in enumerate(data):
    try:
        cursor.execute('INSERT INTO test VALUES(?)', row)
        success_count += 1
    except sqlite3.IntegrityError as ie:
        print(f'第{idx+1}行（数据：{row}）触发异常: {ie}')

print(f'{success_count} 行数据插入成功')
conn.commit()

运行后会直接输出：

第4行（数据：('Larry',)）触发异常: UNIQUE constraint failed: test.name
3 行数据插入成功

这个方法的好处就是逻辑简单，一眼就能看懂，定位错误行也精准；唯一的缺点就是数据量大的时候，逐行插入的效率会比批量插入低不少。

方案二：分块批量插入+二分定位（兼顾效率和精准度，适合大数据量） #

如果你的数据量特别大，不想牺牲太多插入效率，可以试试「分块+递归拆分」的思路：先把数据分成若干大块批量插入，哪一块出错了，就把这块拆成更小的子块，重复这个过程，直到定位到具体的错误行。有点像二分查找，既能保证大部分数据用高效的批量插入，又能精准揪出问题行。

给你写个示例代码：

import sqlite3

def insert_with_error_check(cursor, data):
    if not data:
        return 0
    try:
        # 尝试批量插入当前数据块
        cursor.executemany('INSERT INTO test VALUES(?)', data)
        return len(data)
    except sqlite3.IntegrityError:
        # 如果是单个数据，直接定位错误
        if len(data) == 1:
            print(f'触发约束的行：{data[0]}')
            return 0
        # 拆分数据块，递归检查
        mid = len(data) // 2
        left_success = insert_with_error_check(cursor, data[:mid])
        right_success = insert_with_error_check(cursor, data[mid:])
        return left_success + right_success

conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS test(name TEXT,UNIQUE(name))')

data = [('Larry',),('Curly',),('Moe',),('Larry',)]
total_success = insert_with_error_check(cursor, data)
print(f'{total_success} 行数据插入成功')
conn.commit()

运行后会输出：

触发约束的行：('Larry',)
3 行数据插入成功

这个方法平衡了效率和精准度，大数据量下比逐行插入快很多；唯一的小缺点就是代码逻辑稍微复杂一点，不过理解起来也不难。

方案三：提前检查重复数据（从根源避免异常） #

如果你想从源头避免插入时的异常，可以先查询数据库中已存在的记录，提前找出数据列表里的重复项，过滤后再批量插入。

比如这样写：

import sqlite3

conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS test(name TEXT,UNIQUE(name))')
# 模拟数据库中已有的数据
cursor.executemany('INSERT INTO test VALUES(?)', [('Larry',)])
conn.commit()

data = [('Larry',),('Curly',),('Moe',),('Larry',)]

# 查询数据库中已存在的name
cursor.execute('SELECT name FROM test')
existing_names = {row[0] for row in cursor.fetchall()}

# 找出数据列表里的重复项
duplicate_rows = [row for row in data if row[0] in existing_names]
if duplicate_rows:
    print(f'提前检测到重复数据，将跳过这些行：{duplicate_rows}')

# 过滤重复项后批量插入
filtered_data = [row for row in data if row[0] not in existing_names]
cursor.executemany('INSERT INTO test VALUES(?)', filtered_data)
print(f'{cursor.rowcount} 行数据插入成功')
conn.commit()

运行后输出：

提前检测到重复数据，将跳过这些行：[('Larry',), ('Larry',)]
2 行数据插入成功

这个方法能最大化保留批量插入的效率，但要注意：如果有其他进程或线程同时操作这个数据库，可能会出现「检查时数据不存在，插入时已经被其他进程插入」的情况（竞态条件），所以如果是多进程环境，最好还是结合异常处理来兜底。

总结一下：小数据量选逐行插入最省心，大数据量选分块定位，想提前规避异常就用预检查的方式～

备注：内容来源于stack exchange，提问作者DS_London