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 sessionmaker
Base = 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, text
subq = 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 select
ranked_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 不支持，会报错