sqlalchemy 如何写批量插入或忽略(on conflict do nothing)

SQLAlchemy批量插入冲突忽略需按数据库区分:PostgreSQL用insert().on_conflict_do_nothing(index_elements=["id"]),MySQL用prefix_with("IGNORE"),SQLite用prefix_with("OR IGNORE");通用批量插入可用bulk_insert_mappings()或add_all()。

SQLAlchemy 支持批量插入和冲突忽略(ON CONFLICT DO NOTHING),但具体写法取决于你用的是 Core 还是 ORM,以及数据库后端(PostgreSQL 推荐,MySQL 和 SQLite 有替代方案)。

PostgreSQL:用 insert().on_conflict_do_nothing()

这是最直接的方式,仅适用于 PostgreSQL 9.5+,需配合 Core 的 insert 构造:

  • 指定唯一约束字段(如主键或 UNIQUE 索引列)作为冲突判断依据
  • 不能只写列名,要传入 IndexConstraint 或列对象;常用简写是传列名元组给 index_elements

示例:

from sqlalchemy.dialects.postgresql import insert

stmt = insert(User).values([ {"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}, {"id": 3, "name": "Charlie"} ])

冲突时忽略整行(基于主键 id)

do_nothing_stmt = stmt.on_conflict_do_nothing(index_elements=["id"])

session.execute(do_nothing_stmt) session.commit()

MySQL:用 INSERT IGNOREON DUPLICATE KEY UPDATE

SQLAlchemy 不原生支持 INSERT IGNORE,但可通过 prefix_with 实现:

  • INSERT IGNORE 要求表有主键或唯一索引,冲突时静默跳过
  • 也可用 ON DUPLICATE KEY UPDATE 做“存在则更新”,更可控

示例(MySQL):

stmt = insert(User).values([
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
])

INSERT IGNORE INTO user ...

stmt_ignore = stmt.prefix_with("IGNORE")

session.execute(stmt_ignore) session.commit()

SQLite:用 INSERT OR IGNORE

SQLite 支持 INSERT OR IGNORE,同样靠 prefix_with 注入:

stmt = insert(User).values([...])
stmt_ignore = stmt.prefix_with("OR IGNORE")  # 注意:前面不加 INSERT

session.execute(stmt_ignore) se

ssion.commit()

通用 ORM 批量插入(无冲突处理)

如果只是批量插入、不关心冲突,用 session.bulk_insert_mappings()session.add_all() 更高效:

  • bulk_insert_mappings() 绕过 ORM 生命周期,最快,但不触发事件、不返回主键
  • add_all() 走完整 ORM 流程,适合小批量或需后续操作的场景

示例:

# 快速批量(无冲突逻辑)
session.bulk_insert_mappings(User, [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
])

或带 ORM 行为

users = [User(id=1, name="Alice"), User(id=2, name="Bob")] session.add_all(users) session.commit()

不复杂但容易忽略:冲突忽略依赖底层数据库的语法支持,SQLAlchemy 只负责拼装;务必确保目标字段上有唯一约束,否则 ON CONFLICTIGNORE 不生效。