如何在SQLAlchemy中按reexecution编号获取最新记录 | Python

要实现按场景分组并获取每个场景下reexecution编号最大的记录，仅用query.filter无法完成——它仅能过滤行，无法处理分组后的最大值匹配逻辑。以下是两种可行的SQLAlchemy实现方式：

方法1：子查询关联 #

先查询每个场景的最大reexecution值，再关联原表获取对应完整记录：

from sqlalchemy import func, select

# 假设你的模型类为ScenarioRecord，场景标识字段是scene_id
subquery = select(
    ScenarioRecord.scene_id,
    func.max(ScenarioRecord.reexecution).label('max_reexecution')
).group_by(ScenarioRecord.scene_id).subquery()

result = session.query(ScenarioRecord).join(
    subquery,
    (ScenarioRecord.scene_id == subquery.c.scene_id) & 
    (ScenarioRecord.reexecution == subquery.c.max_reexecution)
).all()

方法2：窗口函数（适用于SQLAlchemy 1.4+） #

使用row_number()窗口函数给每个场景内的记录按reexecution降序排序，取排序为1的记录：

from sqlalchemy import func, select
from sqlalchemy.orm import aliased

# 定义窗口：按场景分组，reexecution降序排序
window = func.row_number().over(
    partition_by=ScenarioRecord.scene_id,
    order_by=ScenarioRecord.reexecution.desc()
).label('rn')

# 子查询添加排序编号
subquery = select(ScenarioRecord, window).subquery()
# 取每个场景中排序为1的记录
aliased_record = aliased(ScenarioRecord, subquery)
result = session.query(aliased_record).filter(subquery.c.rn == 1).all()

注意事项 #

• 请根据实际表结构，将scene_id替换为你用来区分不同场景的字段名
• 如果同一场景下存在多条reexecution值相同的最大记录，方法1会返回所有匹配记录，方法2仅返回其中一条（具体取决于数据库的默认排序规则）

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