
SQLAlchemy 中写 ROW_NUMBER() 窗口函数,核心是用 func.row_number() 配合 over() 方法,指定排序(order_by)和可选的分组(partition_by)。
基础写法:全局排序编号
最简单的情况是对整个结果集按某列排序后编号:
from sqlalchemy import funcquery = session.query( User.name, func.row_number().over(order_by=User.created_at).label('rank') ).order_by(User.created_at)
生成 SQL 类似:
SELECT name, ROW_NUMBER() OVER (ORDER BY created_at) AS rank FROM user ORDER BY created_at按分组排序编号(partition by)
比如给每个部门内的员工按入职时间排名:
query = session.query(
User.name,
User.department,
func.row_number().over(
partition_by=User.department,
order_by=User.created_at
).label('dept_rank')
)
partition_by 支持单字段、多字段(传元组或列表),也支持表达式。
在子查询或 CTE 中使用更清晰
窗口函数不能直接用于 WHERE 或 HAVING,常需嵌套。推荐用子查询或 CTE:
- 子查询方式:
subq = session.query(
User.id,
User.name,
func.row_number().over(order_by=User.score.desc()).label('rn')
).subquery()
query = session.query(subq.c.name).filter(subq.c.rn <= 3) # 取前3名
- CTE 方式(更易读,尤其复杂逻辑):
from sqlalchemy import select, textranked_cte = select( User.name, User.score, func.row_number().over(order_by=User.score.desc()).label('rn') ).cte('ranked')
query = select(ranked_cte.c.name).where(ranked_cte.c.rn <= 3)
注意点和常见坑
-
over()内必须有order_by,否则 SQLAlchemy 会报错(标准 SQL 也要求) - 别名要用
.label()显式定义,否则返回结果中列名可能不明确 - 如果想实现“并列不跳号”(如 RANK())或“并列跳号”(DENSE_RANK()),换用
func.rank()或func.dense_rank() - SQLite 3.25+、PostgreSQL、MySQL 8.0+ 支持窗口函数;旧版 MySQL 或 SQLite 不支持,会报错










