SQLAlchemy使用INSERT IGNORE插入重复数据报错如何解决

• 你通过before_execute事件给INSERT语句追加IGNORE关键字的修改仅作用于SQL执行层，SQLAlchemy ORM层无法感知该逻辑变更。
• 当触发主键/唯一键冲突、INSERT IGNORE实际未写入任何数据时，ORM层仍默认判定插入操作成功，将你构造的ORM实例标记为持久化状态并加入会话的身份映射（identity_map）。
• 若当前会话中已存在同主键的实例，新实例写入身份映射时就会触发对应的SAWarning，提示同主键对象被替换。
• 后续访问该被错误标记为持久化的实例时，只要触发SQLAlchemy的状态同步逻辑（懒加载、字段刷新、修改后提交等），就会发现实例状态和数据库实际数据不匹配，进而抛出ObjectDeletedError异常。

以下方案可根据你的业务场景选择：

方案1：改用SQLAlchemy Core层执行插入（最稳定） #

直接通过Core构造插入语句执行，不通过ORM的session.add()提交对象，完全规避ORM状态管理问题。需要读取数据时再单独发起查询即可：

# Core层插入，不影响ORM会话状态
insert_stmt = Insert(PipetorDiagnostics).values(id=xxx, field1=xxx, ...)
session.execute(insert_stmt)

# 需要使用对象时再查询
obj = session.get(PipetorDiagnostics, xxx)

方案2：插入完成后主动清理会话中的实例（适配现有封装） #

flush完成后立即将刚插入的实例从会话中移除，避免错误的持久化状态残留。你之前调用expunge无效是因为调用时机不对，需要在flush后、访问实例前执行：

obj = PipetorDiagnostics(**data)
session.add(obj)
session.flush()
# 立即清理会话中的对应实例
session.expunge(obj)

# 后续需要使用时重新查询即可
obj = session.get(PipetorDiagnostics, data['id'])

方案3：改用空更新代替INSERT IGNORE（无状态异常） #

修改你的事件监听器，将INSERT IGNORE改为INSERT ... ON DUPLICATE KEY UPDATE执行无意义更新（比如主键赋值为自身），此时重复插入时数据库返回的影响行数为1，ORM层会认为插入成功，不会出现状态异常：

# 原逻辑替换为
if isinstance(element, Insert) and 'ignore_tables' in conn.info and element.table.name in conn.info['ignore_tables']:
    pk_col = element.table.primary_key.columns[0]
    element = element.on_duplicate_key_update({pk_col.name: pk_col})

方案4：仅屏蔽警告（治标，适合无同会话读取的场景） #

如果确定99%场景不会在同一会话读取插入对象，可直接过滤对应警告：

import warnings
from sqlalchemy.exc import SAWarning
warnings.filterwarnings("ignore", category=SAWarning, message="Identity map already had an identity for")

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