SQLAlchemy使用row_number()实现分组排名方法

目前golang学习网上已经有很多关于文章的文章了，自己在初次阅读这些文章中，也见识到了很多学习思路；那么本文《SQLAlchemy 使用 row_number() 实现分组排名的方法如下：导入必要的模块 首先，需要从 sqlalchemy 和 sqlalchemy.sql 中导入相关函数和表达式。from sqlalchemy import func, Column, Integer, String, create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base定义模型类 假设有一个名为 Sales 的表，包含以下字段：id：主键product_id：产品ID（用于分组）sales_amount：销售金额Base = declarative_base() class Sales(Base): __tablename__ = 'sales' id = Column(Integer, primary_key=True) product_id = Column(Integer) sales_amount = Column(Integer)编写窗口函数查询 使用 func.row_number() 并结合 over() 函数来实现分组排名。这里我们按 product_id 分组，并按 sales_amount 降序排列。 from sqlalchemy import select, func, over # 创建会话 engine = create_engine('sqlite:///example.db') Session = sessionmaker(bind=engine) session = Session() # 构建查询》，也希望能帮助到大家，如果阅读完后真的对你学习文章有帮助，欢迎动动手指，评论留言并分享~

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
<p>Base = declarative_base()</p><p>class User(Base):
<strong>tablename</strong> = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
dept = Column(String)
salary = Column(Integer)</p><h1>构造带 row_number 的查询</h1><p>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()
</p>

生成的 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 <= 3 过滤（因为窗口函数在 WHERE 之后执行）。需嵌套一层：

from sqlalchemy import select, text
<p>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')</p><p>top3_per_dept = session.query(subq.c.name, subq.c.dept, subq.c.salary).filter(subq.c.rank <= 3).all()
</p>

或者用 CTE（更清晰）：

from sqlalchemy import select
<p>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')</p><p>top3 = session.query(ranked_cte.c.name, ranked_cte.c.dept, ranked_cte.c.salary).filter(ranked_cte.c.rank <= 3).all()
</p>

注意点与常见问题

• 字段别名必须用 .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 不支持，会报错

终于介绍完啦！小伙伴们，这篇关于《SQLAlchemy使用row_number()实现分组排名方法》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布文章相关知识，快来关注吧！