MySQL用INSERT IGNORE跳过唯一冲突行并返回实际插入数；PostgreSQL用ON CONFLICT DO NOTHING，需指定index_elements；跨库兼容可先查后插但非原子操作。
MySQL 中用 insert_ignore 实现批量去重插入
MySQL 原生支持 INSERT IGNORE，SQLAlchemy 通过 Insert 构造配合 prefix_with("IGNORE") 可直接映射。关键点是：它跳过违反唯一约束（如主键、UNIQUE 索引）的行，不报错，且能拿到实际插入行数。
实操建议：
- 必须确保目标表已定义主键或唯一索引，否则
IGNORE不生效 - 使用
session.execute()而非session.add_all()，后者无法触发IGNORE行为 - 插入后调用
result.rowcount获取真正插入的记录数（不是“尝试插入”的数量）
from sqlalchemy import insertstmt = insert(MyModel).prefix_with("IGNORE") result = session.execute(stmt, [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}, {"id": 1, "name": "c"}]) print(result.rowcount) # 输出：2（id=1 的第二条被忽略） session.commit()
PostgreSQL 中用 ON CONFLICT DO NOTHING 替代
PostgreSQL 没有 IGNORE，对应的是 ON CONFLICT 子句。SQLAlchemy 2.0+ 原生支持 on_conflict_do_nothing，需指定冲突目标（通常是主键或唯一索引字段）。
常见错误现象：
- 漏写
index_elements，导致 SQL 报错there is no unique or exclusion constraint matching the ON CONFLICT specification - 用错字段名（比如写了列名而非 Python 属性名），实际生成 SQL 时出错
- 在 SQLAlchemy 1.4 中强行用 2.0 语法，会抛
AttributeError
from sqlalchemy.dialects.postgresql import insertstmt = insert(MyModel).on_conflict_do_nothing(index_elements=["id"]) result = session.execute(stmt, [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}, {"id": 1, "name": "c"}]) print(result.rowcount) # 输出：2 session.commit()
跨数据库兼容写法：先查后插（低并发适用）
如果必须兼容多种数据库，或业务要求严格控制“哪些被跳过”，可手动做存在性检查。但注意：这不是原子操作，高并发下仍可能重复插入。
使用场景：
- 数据量小（
- 需要精确知道每条记录是否新增/跳过
- 目标库不支持
IGNORE或ON CONFLICT（如某些 SQLite 版本）
性能影响：N 条数据 → 至少 N 次查询 + 最多 N 次插入，远慢于原生批量忽略。
ids_to_insert = [1, 2, 3]
existing_ids = set(session.scalars(select(MyModel.id).where(MyModel.id.in_(ids_to_insert))).all())
new_records = [MyModel(id=i, name=f"item_{i}") for i in ids_to_insert if i not in existing_ids]
session.add_all(new_records)
session.flush()
print(len(new_records)) # 插入数量
返回插入数量时最容易忽略的细节
rowcount 在不同驱动和模式下行为不一致：
- SQLite 的
pysqlite驱动在 executemany 后可能返回-1，必须用单条execute+ 手动构造多值 INSERT - MySQL 的
pymysql和mysqldb通常返回正确值，但开启autocommit=False时需commit()后才稳定 - PostgreSQL 的
psycopg2在ON CONFLICT下返回准确值，但若用了RETURNING子句，rowcount会失效，得用result.fetchall()数长度
复杂点在于：没有一个写法能在所有数据库、所有驱动、所有 SQLAlchemy 版本上 100% 返回可靠数字。最稳妥的做法是——明确你的目标数据库和驱动，然后选对应方案，别强求“一次写完到处跑”。
