SQLAlchemy：一对多关联下限制联查结果每个项目仅返回首行

你已经找对了方向——用CASE语句定义自定义排序规则是实现需求的关键一步，不过要完成「每个Project仅返回排序后的首行StudentList」这个目标，咱们得结合SQLAlchemy的窗口函数来实现分组标记行号的逻辑。

完整实现代码 #

from sqlalchemy import case, func, over
from sqlalchemy.orm import aliased

# 定义自定义排序权重：APPROVED优先级最高，None最低
_whens = {
    ProjectStatus.APPROVED: 1,
    ProjectStatus.REJECTED: 2,
    ProjectStatus.SUBMITTED: 3,
    None: 4
}
# 生成CASE表达式，将status转换为排序用的数值
sort_order = case(value=StudentList.student_list_status_id, whens=_whens)

# 使用窗口函数ROW_NUMBER()，按Project分组、自定义排序标记行号
row_num = func.row_number().over(
    partition_by=StudentList.project_id,  # 按所属Project分组
    order_by=sort_order.asc()  # 按权重升序，数值越小优先级越高
).label('row_num')

# 子查询：获取所有带行号的StudentList记录
subquery = self._session.query(StudentList, row_num).subquery()

# 别名化子查询中的StudentList，方便后续查询操作
aliased_sl = aliased(StudentList, subquery)

# 最终查询：筛选每个Project中行号为1的记录（即排序后的首条）
result = self._session.query(aliased_sl).filter(subquery.c.row_num == 1).all()

return result

代码解释 #

1. 自定义排序规则：通过case函数把student_list_status_id映射成你需要的优先级数值，这样就能按照APPROVED > REJECTED > SUBMITTED > None的顺序排序。
2. 窗口函数分组标记：row_number().over()会给每个Project下的StudentList按自定义排序分配行号，组内第一条记录的行号为1。
3. 筛选首行记录：最后通过过滤行号等于1的条件，就能拿到每个Project排序后的首条StudentList。

关联Project实体查询（可选） #

如果需要同时获取Project的信息，可以调整查询语句，关联Project表：

result = (
    self._session.query(Project, aliased_sl)
    .join(aliased_sl, Project.project_id == aliased_sl.project_id)
    .filter(subquery.c.row_num == 1)
    .all()
)

这样返回的结果就是每个Project与其对应的首条StudentList的配对数据。

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