sqlalchemy 如何写“批量删除”并返回删除行数

SQLAlchemy 2.0+ 中 delete() 执行后 rowcount 不可靠,准确获取删除行数需分场景:PostgreSQL 用 RETURNING + CTE;通用方案用 session.connection().execute() 原生 SQL 并读取 rowcount。

SQLAlchemy 2.0+ 的 delete() 语句不支持直接返回行数

直接执行 session.execute(delete(User).where(User.status == 'inactive')) 后调用 result.rowcount,在大多数数据库后端(如 PostgreSQL、SQLite)会返回 -1 或不可靠值。这是因为 SQLAlchemy 默认使用“无结果集”的执行模式,底层未触发 cursor.rowcount 的正确读取时机。

真正能拿到准确删除行数的方式,是让数据库执行带 RETURNING(或等效机制)的语句,并显式 fetch —— 但标准 delete() 不提供该

能力,必须换路径。

select().with_for_update() + delete() 分两步确保行数准确

适用于需要强一致性、且不能接受“先查再删”中间被并发修改的场景。本质是加锁查出待删主键,再按主键删:

  • 先执行 session.execute(select(User.id).where(User.status == 'inactive').with_for_update()),获取所有匹配的 id 列表
  • 记录列表长度,即为将要删除的行数
  • 再执行 session.execute(delete(User).where(User.id.in_(id_list)))
  • 最后 session.commit()

注意:id_list 不能过大(如超 10 万),否则 IN 子句可能触发数据库限制;此时应分批处理。

PostgreSQL 可用 RETURNING + cte() 实现单语句获取行数

PostgreSQL 支持在 DELETE 中用 RETURNING 返回被删行,结合 CTE 可绕过 ORM 层限制:

from sqlalchemy import delete, select, func
from sqlalchemy.dialects.postgresql import insert

构造 CTE:先锁定并标记待删行

del_cte = ( delete(User) .where(User.status == 'inactive') .returning(User.id) .cte('deleted_rows') )

统计 CTE 中返回的行数

stmt = select(func.count()).select_from(del_cte)

result = session.execute(stmt).scalar()

result 就是真实删除行数

session.commit()

⚠️ 这个方案仅限 PostgreSQL;SQLite 不支持 RETURNINGDELETE 中,MySQL 8.0+ 虽支持 RETURNING,但 SQLAlchemy 当前(2.0.30)尚未适配其 delete().returning() 语法。

最通用稳妥的做法:用原生 SQL + connection.execute()

绕过 ORM 缓存和语句重写,直连底层 Connection,保证 rowcount 可用:

  • 对 SQLite / PostgreSQL / MySQL 均有效
  • 需手动处理参数化防止 SQL 注入:text("DELETE FROM users WHERE status = :status")
  • 执行后立即读 result.rowcount,不要等 commit

示例:

from sqlalchemy import text

stmt = text("DELETE FROM users WHERE status = :status") result = session.connection().execute(stmt, {"status": "inactive"}) deleted_count = result.rowcount session.commit() # 注意:connection.execute 不自动参与 session 事务,需显式 commit

关键点:必须用 session.connection().execute(),而非 session.execute();后者走 ORM 执行器,rowcount 行为不可靠。

实际项目里,如果只是统计用途且允许轻微误差,“先查 ID 数再删”最易懂;如果必须精确且数据库是 PostgreSQL,优先用 CTE + RETURNING;其余情况,原生 SQL 是唯一跨库稳态解法。