如何在Django中按content_type_id筛选每个分类下最新4条记录？

我来给你捋捋怎么搞定这个需求——PostgreSQL的窗口函数ROW_NUMBER()刚好能完美解决你的问题，它可以在单查询里完成分组排序并筛选Top N，完全适配你的要求，还能和Django-Filter无缝结合。

核心思路 #

我们可以借助Window函数，按content_type_id进行分区（相当于分组），然后对每个分区内的记录按access_time降序排序，给每条记录分配一个行号。最后筛选出行号≤4的记录，就能得到每个分类下的最新4条数据。

具体代码实现 #

1. 基础查询逻辑 #

先写出核心的查询代码，直接实现分组取Top4的效果：

from django.db.models import Window, F
from django.db.models.functions import RowNumber

# 给每条记录添加分组内的行号，再筛选行号≤4的记录
queryset = EntriesChangeLog.objects.annotate(
    row_num=Window(
        expression=RowNumber(),
        partition_by=F('content_type_id'),  # 按content_type_id分组
        order_by=F('access_time').desc()    # 组内按access_time倒序排
    )
).filter(row_num__lte=4)

这个查询是单语句执行的，不管你的数据集包含1种、2种还是全部3种分类，都能自动为每个分类保留最新的4条记录。

2. 适配Django-Filter #

如果要把这个逻辑整合到Django-Filter的FilterSet里，只需要在get_queryset方法中注入窗口函数和筛选逻辑即可：

import django_filters

class EntriesChangeLogFilter(django_filters.FilterSet):
    # 这里定义你的自定义过滤字段，比如用户、操作类型等
    user = django_filters.ModelChoiceFilter(queryset=get_user_model().objects.all())
    operation_type = django_filters.ChoiceFilter(choices=OperationTypeChoices.choices)

    class Meta:
        model = EntriesChangeLog
        fields = ['user', 'operation_type', 'content_type']

    def get_queryset(self):
        # 先获取过滤后的基础查询集
        queryset = super().get_queryset()
        # 添加窗口函数注解并筛选每个分类的Top4
        queryset = queryset.annotate(
            row_num=Window(
                expression=RowNumber(),
                partition_by=F('content_type_id'),
                order_by=F('access_time').desc()
            )
        ).filter(row_num__lte=4)
        return queryset

这样当你使用这个FilterSet进行过滤时，所有返回的结果都会自动按content_type_id分组，保留每组最新的4条记录。

为什么不用你之前的Subquery方案？ #

你之前实现的取单条最新记录的方法，是通过子查询获取每个分组的最大时间再过滤，但这种思路没法直接扩展到取Top N——一来同一个分组可能有多个记录的access_time相同，二来没法直接关联出前N条数据。而窗口函数天生就是为这种「分组内排序取Top N」的场景设计的，逻辑更清晰，性能也更优（单查询完成，不需要多次关联）。

注意事项 #

• 这个方案依赖PostgreSQL的窗口函数支持，你用的Postgres 12完全兼容；
• 如果需要对最终结果进行全局排序（比如整体按access_time降序），直接在查询末尾追加.order_by('-access_time')即可，不会影响分组筛选的结果；
• 如果你需要针对特定用户筛选（比如你之前代码里的user=OuterRef('user')），只需要在基础查询或FilterSet的过滤条件中添加user=xxx的限制，窗口函数会自动在过滤后的数据集上进行分组处理。

内容的提问来源于stack exchange，提问作者Aleksei Khatkevich