如何使用row_number()实现SQLAlchemy的query.delete()操作

方法1：通过EXISTS子查询动态匹配所有列 #

如果不想硬编码任何列名，可以动态获取模型的所有列，用EXISTS子查询匹配子查询中的Foo行：

from sqlalchemy import exists, and_

# 生成带行号的子查询
subquery = session.query(
    Foo,
    func.row_number().over(order_by=desc(Foo.bar)).label("row_number")
).subquery()

# 过滤行号大于3的结果
filtered_sub = session.query(subquery.c).filter(subquery.c.row_number > 3)

# 构建删除查询，用EXISTS匹配所有列
delete_query = session.query(Foo).filter(
    exists().where(
        and_(*[
            getattr(Foo, col_name) == getattr(filtered_sub.c, col_name)
            for col_name in Foo.__table__.columns.keys()
        ])
    )
)

# 执行删除
delete_query.delete(synchronize_session=False)
session.commit()

方法2：用CTE简化逻辑（推荐） #

CTE（公共表表达式）的写法更清晰，能避免多层子查询嵌套：

from sqlalchemy import func, desc, exists, and_

# 定义CTE，包含Foo所有字段和行号
foo_cte = session.query(
    Foo,
    func.row_number().over(order_by=desc(Foo.bar)).label("row_number")
).cte()

# 构建DELETE语句，通过EXISTS关联CTE中的目标行
delete_stmt = Foo.__table__.delete().where(
    exists().where(
        and_(*[
            getattr(Foo, col) == getattr(foo_cte.c, col)
            for col in Foo.__table__.columns.keys()
        ])
    ).where(foo_cte.c.row_number > 3)
)

# 执行删除
session.execute(delete_stmt)
session.commit()

为什么你原来的写法不行？ #

Foo.in_(subquery)是错误用法——in_()是列对象的方法，不是模型类的方法。模型类本身不能直接用于in_条件，必须针对具体列或者用EXISTS来匹配整行数据。

优化建议：用主键关联（高效且避免全列匹配） #

如果你的模型有主键（几乎所有业务模型都应该有），可以动态获取主键列来关联，比全列匹配更高效：

from sqlalchemy import func, desc

# 动态获取主键列名
pk_col_name = Foo.__table__.primary_key.columns.keys()[0]
pk_col = getattr(Foo, pk_col_name)

# 定义CTE，只取主键和行号
foo_cte = session.query(
    pk_col,
    func.row_number().over(order_by=desc(Foo.bar)).label("row_number")
).cte()

# 构建DELETE语句
delete_stmt = Foo.__table__.delete().where(
    pk_col.in_(
        session.query(foo_cte.c[pk_col_name]).filter(foo_cte.c.row_number > 3)
    )
)

session.execute(delete_stmt)
session.commit()

这种方法既避免了硬编码ID，又利用主键索引提升了删除效率。

内容的提问来源于stack exchange，提问作者Evgenii Kozlov