sqlalchemy 如何写“窗口函数”里的 row_number() 排名

13次阅读

sqlalchemy 如何写“窗口函数”里的 row_number() 排名

sqlAlchemy 中写 ROW_NUMBER() 窗口函数,核心是用 func.row_number() 配合 over() 方法,指定排序(order_by)和可选的分组(partition_by)。

基础写法:全局排序编号

最简单的情况是对整个结果集按某列排序后编号:

from sqlalchemy import func 

query = 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 中使用更清晰

窗口函数不能直接用于 WHEREHAVING,常需嵌套。推荐用子查询或 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, text 

ranked_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+、postgresqlmysql 8.0+ 支持窗口函数;旧版 MySQL 或 SQLite 不支持,会报错

text=ZqhQzanResources