能否使用SQLAlchemy的DBSession.bulk_save_objects批量插入并忽略重复键

当然可以做到！不过DBSession.bulk_save_objects本身并没有内置的参数来直接忽略重复键导致的完整性错误——默认情况下，只要有一条记录触发重复键冲突，整个批量操作就会抛出IntegrityError并回滚。下面分享两种可行的实现思路，其中第一种是效率更高的推荐方案：

方法1：结合数据库原生语法与SQLAlchemy insert()构造器（推荐） #

这种方式直接利用数据库的原生冲突处理逻辑，不需要在应用层做复杂的错误捕获，效率最高。

针对PostgreSQL（支持ON CONFLICT语法） #

如果你使用PostgreSQL，可以借助SQLAlchemy的on_conflict_do_nothing()方法，指定唯一键字段来忽略冲突：

from sqlalchemy import insert

# 假设你的模型是User，唯一约束字段为email
insert_stmt = insert(User).values([
    {"name": "Alice", "email": "alice@example.com"},
    {"name": "Bob", "email": "bob@example.com"},
    # 这条记录email重复，会被自动忽略
    {"name": "DuplicateAlice", "email": "alice@example.com"}
]).on_conflict_do_nothing(index_elements=["email"])

DBSession.execute(insert_stmt)
DBSession.commit()

针对MySQL（支持INSERT IGNORE语法） #

对于MySQL，可以通过prefix_with('IGNORE')给插入语句添加INSERT IGNORE前缀，自动跳过重复键记录：

from sqlalchemy import insert

insert_stmt = insert(User).values([
    {"name": "Alice", "email": "alice@example.com"},
    {"name": "Bob", "email": "bob@example.com"},
    {"name": "DuplicateAlice", "email": "alice@example.com"}
]).prefix_with('IGNORE')

DBSession.execute(insert_stmt)
DBSession.commit()

方法2：使用bulk_save_objects并处理完整性错误（不推荐，效率较低） #

如果你一定要用bulk_save_objects，可以通过分块插入+捕获错误的方式实现，但这种方式效率不高，因为每次冲突都会触发回滚，需要额外的重试逻辑：

逐行插入（简单但低效） #

from sqlalchemy.exc import IntegrityError

# 待插入的对象列表
objects_to_add = [
    User(name="Alice", email="alice@example.com"),
    User(name="Bob", email="bob@example.com"),
    User(name="DuplicateAlice", email="alice@example.com")
]

for obj in objects_to_add:
    try:
        DBSession.bulk_save_objects([obj])
        DBSession.commit()
    except IntegrityError:
        DBSession.rollback()
        # 跳过重复记录，继续插入下一条
        continue

分块插入+逐行重试（相对高效一点） #

如果数据量较大，可以先分块批量插入，遇到冲突时再逐行处理冲突块：

from sqlalchemy.exc import IntegrityError

def bulk_insert_ignore_duplicates(session, model, objects, chunk_size=100):
    for i in range(0, len(objects), chunk_size):
        chunk = objects[i:i+chunk_size]
        try:
            session.bulk_save_objects(chunk)
            session.commit()
        except IntegrityError:
            session.rollback()
            # 逐行插入当前块，跳过冲突记录
            for obj in chunk:
                try:
                    session.bulk_save_objects([obj])
                    session.commit()
                except IntegrityError:
                    session.rollback()
                    continue

# 调用示例
bulk_insert_ignore_duplicates(DBSession, User, objects_to_add)

为什么优先推荐方法1？ #

• 性能更优：数据库层面的冲突处理是原子操作，不需要应用层和数据库多次交互，速度更快。
• 代码更简洁：不需要复杂的错误捕获和回滚逻辑，可读性更强。
• 可靠性更高：避免了分块/逐行插入可能带来的性能损耗和潜在的数据一致性问题。

需要注意的是，方法1要求你的数据库支持对应的冲突处理语法（比如PostgreSQL的ON CONFLICT、MySQL的INSERT IGNORE），并且你的模型已经正确定义了唯一键约束。

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