SQLAlchemy中用func.row_number().over(partition_by=..., order_by=...)实现分组排名，需用label()命名、嵌套子查询或CTE过滤Top-N，并注意NULL排序及数据库兼容性。
SQLAlchemy 中用 func.row_number() 配合 over() 可以实现分组排名，关键在于正确构造 over(partition_by=..., order_by=...) 子句。
基础写法：单字段分组 + 单字段排序
假设有一个用户表 users，想按部门（dept）分组，按薪资（salary）降序排，给每组内用户打排名：
from sqlalchemy import create_engine, Column, Integer, String, func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmakerBase = declarative_base()
class User(Base): tablename = 'users' id = Column(Integer, primary_key=True) name = Column(String) dept = Column(String) salary = Column(Integer)
构造带 row_number 的查询
ranked = session.query( User.name, User.dept, User.salary, func.row_number().over( partition_by=User.dept, order_by=User.salary.desc() ).label('rank') ).all()
生成的 SQL 类似：
SELECT name, dept, salary, ROW_NUMBER() OVER (PARTITION BY dept ORDER BY salary DESC) AS rank FROM users;
支持多字段分组和排序
若需按多个字段分组（如部门 + 岗位），或按多个字段排序（先按薪资降序、再按入职时间升序）：
-
多字段分组：传入元组或列表，如
partition_by=[User.dept, User.role] -
多字段排序：用逗号连接多个
order_by字段，或传入列表，如order_by=[User.salary.desc(), User.hire_date.asc()]
示例：
func.row_number().over(
partition_by=[User.dept, User.role],
order_by=[User.salary.desc(), User.hire_date.asc()]
).label('rank')
在子查询或 CTE 中使用（推荐用于过滤 Top-N）
直接在主查询中加 row_number() 无法用 WHERE rank 过滤（因为窗口函数在 WHERE 之后执行）。需嵌套一层：
from sqlalchemy import select, textsubq = select( User.name, User.dept, User.salary, func.row_number().over( partition_by=User.dept, order_by=User.salary.desc() ).label('rank') ).subquery('ranked')
top3_per_dept = session.query(subq.c.name, subq.c.dept, subq.c.salary).filter(subq.c.rank <= 3).all()
或者用 CTE（更清晰）：
from sqlalchemy import selectranked_cte = select( User.name, User.dept, User.salary, func.row_number().over( partition_by=User.dept, order_by=User.salary.desc() ).label('rank') ).cte('ranked')
top3 = session.query(ranked_cte.c.name, ranked_cte.c.dept, ranked_cte.c.salary).filter(ranked_cte.c.rank <= 3).all()
注意点与常见问题
-
字段别名必须用
.label()，否则 ORM 查询可能无法映射结果列 -
排序字段不能为 None 或 NULL：默认 NULL 排最前，如需 NULL 排最后，用
User.salary.desc().nulls_last() -
区分 row_number / rank / dense_rank：SQLAlchemy 同样支持
func.rank()和func.dense_rank()，语义不同（是否跳过重复名次） - 数据库兼容性：窗口函数在 PostgreSQL、SQL Server、Oracle、SQLite 3.25+、MySQL 8.0+ 中可用；旧版 MySQL 或 SQLite 不支持，会报错
