SQLAlchemy子查询Limit与Order_by：FastAPI关联查询优化问题

你之前的查询逻辑存在核心问题：limit(number_of_commerces)是对全局结果做限制，而非按每个子分类分组后取前N条，因此无法实现每个子分类下返回指定数量商家的需求。

下面是针对多对多场景的正确实现方案，利用SQL窗口函数ROW_NUMBER()对每个子分类下的商家单独编号排序，再筛选符合条件的记录：

from sqlalchemy import func, over

@classmethod
def get_subcategory_commerce(cls, id: int, db: Session, number_of_commerces: int = 5):
    # 子查询：给每个子分类下的商家按importance降序生成行号
    commerce_subq = db.query(
        Commerce,
        Subcategory.id.label("subcategory_id"),
        func.row_number().over(
            partition_by=Subcategory.id,  # 按子分类ID分组
            order_by=Commerce.importance.desc()  # 组内按importance降序排序
        ).label("row_num")
    ).join(Commerce.subcategories).subquery()

    # 主查询：获取指定分类下的子分类，并关联过滤后的商家（仅保留每个子分类前N条）
    subcategories = db.query(Subcategory)\
        .filter(Subcategory.main_category_id == id)\
        .outerjoin(
            commerce_subq,
            (Subcategory.id == commerce_subq.c.subcategory_id) & (commerce_subq.c.row_num <= number_of_commerces)
        )\
        .options(contains_eager(Subcategory.commerces, alias=commerce_subq))\
        .order_by(Subcategory.id, commerce_subq.c.importance.desc())\
        .all()
    
    return subcategories

逻辑说明： #

1. 窗口函数分组编号：通过partition_by=Subcategory.id将商家按所属子分类分组，每组内按importance降序生成行号（从1开始）
2. 筛选Top N记录：主查询关联子查询时，通过row_num <= number_of_commerces只保留每个子分类下的前N条商家
3. 避免N+1查询：使用contains_eager直接加载过滤后的商家集合，替代默认的懒加载，提升查询效率

注意事项： #

• 确保你的Subcategory和Commerce模型已正确定义多对多关系，例如：

  # 多对多关联表
  subcategory_commerce_association = Table(
      "subcategory_commerce",
      Base.metadata,
      Column("subcategory_id", Integer, ForeignKey("subcategory.id")),
      Column("commerce_id", Integer, ForeignKey("commerce.id"))
  )
  
  # Subcategory模型内的关联字段
  commerces = relationship("Commerce", secondary=subcategory_commerce_association, back_populates="subcategories")
  

• 你的ShowSubcategoryCommerce Pydantic模型已经开启orm_mode，可以直接将SQLAlchemy模型实例转换为响应格式

内容的提问来源于stack exchange，提问作者Juan Cotrino