如何在单查询中按分类获取指定数量的关联表记录？

问题背景 #

现有三张数据表：

• article_page：存储文章数据，对应Django模型Page
• article_page_category：article_page与article_category的多对多关联中间表
• article_category：存储分类信息，对应Django模型Category

需求是通过单次数据库查询，从每个分类中获取指定数量（10-20条）的文章记录（例如business、technology、sports三类各取10条），避免因分类数量多导致多次查询的高成本。当前使用子查询的方案仅能获取单条文章ID，不符合需求且效率不佳。

解决方案 #

方法1：MySQL窗口函数实现（推荐，MySQL 8.0+） #

MySQL 8.0及以上支持ROW_NUMBER()窗口函数，可在数据库层面完成分组排序和前N条筛选，仅需一次查询：

from django.db.models import Window, F
from django.db.models.functions import RowNumber
from rest_framework.response import Response

class GroupedArticleListView(APIView):
    def get(self, request):
        # 定义窗口：按分类分组，文章按日期倒序排序
        rank_window = Window(
            partition_by=F('category'),
            order_by=F('date').desc()
        )
        
        # 为文章标注组内行号，筛选每个分类前10条
        top_articles = Page.objects.filter(
            category__taxon=1
        ).annotate(
            row_num=RowNumber().over(rank_window)
        ).filter(row_num__lte=10)
        
        # 按分类分组整理返回结果
        grouped_data = {}
        for article in top_articles:
            cat_name = article.category.name
            grouped_data.setdefault(cat_name, []).append({
                'id': article.id,
                'title': article.title,
                'date': article.date,
                # 按需添加其他字段
            })
        
        return Response(grouped_data)

方法2：原生SQL兼容低版本MySQL（MySQL 5.x） #

如果使用不支持窗口函数的MySQL 5.x，可通过变量计数实现分组筛选：

from rest_framework.response import Response

class GroupedArticleListView(APIView):
    def get(self, request):
        limit = 10
        # 基础查询：关联表并按分类、日期排序
        base_sql = """
        SELECT ap.*, ac.name AS category_name
        FROM article_page ap
        JOIN article_page_category apc ON ap.id = apc.article_page_id
        JOIN article_category ac ON apc.article_category_id = ac.id
        WHERE ac.taxon = 1
        ORDER BY ac.id, ap.date DESC
        """
        # 嵌套查询添加行号筛选
        filtered_sql = f"""
        SELECT * FROM (
            SELECT *,
                   @row := IF(@current_cat = category_name, @row + 1, 1) AS row_num,
                   @current_cat := category_name
            FROM ({base_sql}) AS sorted_articles,
                 (SELECT @row := 0, @current_cat := '') AS init_vars
        ) AS ranked_articles
        WHERE row_num <= {limit}
        """
        
        # 执行原生查询并整理结果
        articles = Page.objects.raw(filtered_sql)
        grouped_data = {}
        for art in articles:
            grouped_data.setdefault(art.category_name, []).append({
                'id': art.id,
                'title': art.title,
                'date': art.date,
            })
        
        return Response(grouped_data)

现有方案的问题 #

你当前的代码通过Subquery结合first()仅能获取每个分类的第一条文章ID，无法满足取10条的需求；且子查询会为每个分类单独执行一次，本质是N+1查询，分类数量多时效率极低。

内容的提问来源于stack exchange，提问作者Steven Matthews