如何在Django中实现帖子Top3热门反应的嵌套查询？

背景环境 #

使用Django搭配MySQL数据库，已定义如下Model：

class Post(models.Model):
    title = models.CharField(max_length=255, db_index=True)
    .
    .
    .
    author = models.ForeignKey('Account', on_delete=models.SET_NULL, null=True )
    tags = models.ManyToManyField('Tag', blank=True)
    reactions = models.ManyToManyField('Reaction', through="PostReaction", related_name='reactions', blank=True)

class Account(AbstractUser):
    slug = models.SlugField(max_length=150, db_index=True, blank=True, null=True)
    avatar = models.URLField(default='', blank=True)

class PostReaction(models.Model):
    reaction = models.ForeignKey('Reaction', on_delete=models.CASCADE, )
    post = models.ForeignKey('Post', on_delete=models.CASCADE,)
    reacted_at = models.DateTimeField(auto_now_add=True)
    author = models.ForeignKey('Account', on_delete=models.CASCADE, )

class Reaction(models.Model):
    reaction_name = models.CharField(max_length=250, unique=True)
    reaction_url = models.CharField(max_length=2048)
    reaction_category = models.ForeignKey('ReactionCategory', on_delete=models.CASCADE)

已实现展示已发布帖子列表的View：

class PostListView(generics.ListAPIView):

    queryset= Post.objects.filter(is_publish = True).select_related('author').prefetch_related('tags','reactions')    

    serializer_class = PostListReadSerializer

以及用于生成帖子JSON列表的序列化器：

class TagListSerializer(serializers.ModelSerializer):
    class Meta:
        model = Tag
        fields = '__all__'

class PostShortReactionSerializer(serializers.ModelSerializer):
    class Meta:
        model = Reaction
        fields = ["reaction_name", "reaction_url", "id"]

class PostAccountSerializer(serializers.ModelSerializer):
    class Meta:
        model = Account
        fields = ["username", "avatar"]

class PostListReadSerializer(serializers.ModelSerializer):
    tags = TagListSerializer(many=True, read_only=True)
    reactions = PostShortReactionSerializer(many=True, read_only=True)
    author = PostAccountSerializer(read_only=True)

    class Meta:
        model = Post
        fields='__all__'

需求与问题 #

需求：在返回的JSON响应中，每个帖子的Reaction字段需包含该帖子的Top3反应，按该反应的使用数量降序排序。预期输出结构如下：

{
  "id": 68,
  "tags": [
            {
             "id": 1,
             "name": "Xxx",
             "slug": "xxx"
             }...
            ],
   "reactions": [
                  {
                   "reaction_name": "xxx",
                    "reaction_url": "xxx",
                    "id": 1
                },
                {
                    "reaction_name": "xxx2",
                    "reaction_url": "xxx2",
                    "id": 2
                },
                {
                    "reaction_name": "xxx3",
                    "reaction_url": "xxx3",
                    "id":3
                }
   ],
   "author": {
               "username": "xxx",
               "avatar": "xxx"
             },
  *....Other post's data....*           
 }

已尝试两种子查询方案但均未达成目标：

方案一 #

sq= Subquery(PostReaction.objects.filter(post=OuterRef('id')).values_list('id', flat=True)[:3]) 

subquery = PostReaction.objects.filter(post=68).values('reaction__id').annotate(reaction_count=Count('reaction__id')).order_by('-reaction_count')[:3]

queryset = Post.objects.filter(is_publish = True).select_related('author').prefetch_related('tags', Prefetch('reactions', queryset=PostReaction.objects.filter(post=sq)))

该方案构造的子查询未在输出中生效。

方案二 #

queryset= Post.objects.filter(is_publish = True).select_related('author').prefetch_related('tags','reactions').annotate(reactions_count = Count('reactions__id')).order_by('-reactions_count')   

该方案分组失败，添加[:3]会对全局查询结果切片，而非为每个帖子单独获取Top3反应。

现寻求正确的实现方案，以满足每个帖子返回Top3热门反应的需求。

解决方案 #

要实现每个帖子返回Top3热门反应，需要结合子查询统计反应次数+自定义Prefetch获取排序后的Top3反应，具体步骤如下：

1. 构造统计每个帖子反应次数的子查询 #

首先统计每个帖子下各Reaction的使用次数，用Subquery和annotate实现：

from django.db.models import Subquery, OuterRef, Count, Q

# 子查询：统计每个帖子下各reaction的使用次数，按次数降序
reaction_count_subquery = PostReaction.objects.filter(
    post=OuterRef('pk'),
    reaction=OuterRef('reaction__pk')
).values('reaction').annotate(
    count=Count('id')
).values('count')

2. 自定义Prefetch查询集，获取每个帖子的Top3反应 #

为每个帖子的reactions关联字段，自定义Prefetch的查询集，筛选出该帖子下次数最多的前3个Reaction：

# 构造每个帖子的Top3反应查询集
top_reactions_queryset = Reaction.objects.filter(
    postreaction__post=OuterRef('pk')
).annotate(
    reaction_count=Subquery(reaction_count_subquery)
).order_by('-reaction_count').distinct()[:3]

# 使用Prefetch关联这个查询集
prefetch_reactions = Prefetch(
    'reactions',
    queryset=top_reactions_queryset,
    to_attr='top_reactions'
)

3. 更新View的queryset #

修改PostListView的queryset，替换原来的prefetch_related为自定义的Prefetch：

class PostListView(generics.ListAPIView):
    queryset= Post.objects.filter(is_publish = True).select_related('author').prefetch_related(
        'tags',
        prefetch_reactions
    )
    serializer_class = PostListReadSerializer

4. 修改序列化器，输出Top3反应 #

修改PostListReadSerializer，将原来的reactions字段替换为预取的top_reactions：

class PostListReadSerializer(serializers.ModelSerializer):
    tags = TagListSerializer(many=True, read_only=True)
    reactions = PostShortReactionSerializer(source='top_reactions', many=True, read_only=True)
    author = PostAccountSerializer(read_only=True)

    class Meta:
        model = Post
        fields='__all__'

关键说明 #

• 通过Subquery为每个Reaction计算在对应帖子中的使用次数，确保排序基于每个帖子的本地统计，而非全局。
• 使用Prefetch的to_attr参数将结果存储到自定义属性top_reactions，避免覆盖原有的reactions关联数据。
• 适配MySQL对LIMIT在子查询中的支持限制，确保每个帖子单独获取Top3反应，不会出现全局切片问题。

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