如何在SQLAlchemy中实现部分回滚以跳过异常SQL语句？

解决方法 #

你可以根据自己的技术栈从以下3种成熟方案中选择：

方案1：嵌套事务（Savepoint）实现部分回滚 #

SQLAlchemy支持通过创建保存点实现单条语句粒度的回滚，不会撤销已处理的其他操作，完全符合你不想提前提交、统一追踪进度的需求，示例代码如下：

from sqlalchemy.exc import IntegrityError

# parsed_orm_objs为你解析得到的所有ORM对象列表
for obj in parsed_orm_objs:
    session.add(obj)
    # 创建保存点
    nested_txn = session.begin_nested()
    try:
        session.flush()
        nested_txn.commit()
    except IntegrityError:
        # 仅回滚到当前保存点，不影响之前的操作
        nested_txn.rollback()
        # 将报错对象从会话中移除，避免后续提交再次触发错误
        session.expunge(obj)
        # 可在此处添加重复记录的日志逻辑
        continue

# 所有语句处理完成后统一提交
session.commit()

该方案的优势是依赖数据库唯一约束保证正确性，没有并发写入的竞态问题，不需要额外的查询请求，性能优于前置校验。唯一需要注意的是必须将触发错误的ORM对象从当前会话中清理干净。

方案2：使用数据库原生Upsert语法跳过重复记录 #

如果你的底层数据库支持冲突忽略语法（PostgreSQL、MySQL 8.0+、SQLite 3.24+均支持），可以直接构造跳过重复的插入语句，连异常捕获逻辑都不需要，是性能最优的方案，PostgreSQL下的示例如下：

from sqlalchemy.dialects.postgresql import insert

for data in parsed_insert_data:
    stmt = insert(你的ORM模型).values(**data).on_conflict_do_nothing(
        index_elements=[你的ORM模型.复合主键字段1, 你的ORM模型.复合主键字段2]
    )
    session.execute(stmt)

session.commit()

该方案的优势是逻辑最简、性能最高，缺点是和具体数据库方言绑定，多数据库兼容场景下需要额外适配。

方案3：前置校验查询 #

即每次插入前先查询对应复合主键的记录是否存在，仅当不存在时才加入会话。该方案逻辑直观、无数据库绑定，但有两个明显缺陷：一是每条插入需要多一次查询请求，性能较差；二是多进程并发写入场景下，查询和插入的间隙可能有新数据写入，还是会触发唯一约束报错，需要额外加兜底的异常处理逻辑，因此不推荐单独使用。

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