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

苏承栈

2026-10-25 • 👁 0人浏览过

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()

                                                        
                                                                
                                                                
                                                                        MySQL（Linux）
                                                                        MySQL 9.6.0是面向Linux平台的2026年创新版本，核心架构迎来重大革新。其将外键约束与级联操作从InnoDB引擎层上移至SQL层，确保所有数据变更均被完整记录至Binlog，彻底解决了CDC（变更数据捕获）与主从复制中的数据不一致难题。此外，该版本引入container_aware启动选项以原生适配容器环境，并对审计日志进行了组件化重构，为追求极致数据一致性与云原生体验的开发者提供了全新选择。
                                                                
                                                                下载 
                                                        
                                                
result 就是真实删除行数

session.commit()

⚠️ 这个方案仅限 PostgreSQL；SQLite 不支持

RETURNING

在

DELETE

中，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 是唯一跨库稳态解法。

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

统计 CTE 中返回的行数

result 就是真实删除行数

相关文章