[Django REST framework] ModelViewSet response に集計項目 (annotate) を追加する

概要

• ModelViewSet に annotate で集計した項目を追加する
  • ModelViewSet の queryset で annotate する
  • Serializer で read_only=True な field を追加する
    • annotate で指定した名称と Serializer の field 名を一致させる
• ついでに、対象の Model と集計する対象の Model にリレーションが無い例で試してみる
• Django 4.2 で動作確認済み

サンプルプロジェクト

• ブログシステム
• 下書き (DraftPost) と、公開済み記事 (PublishedPost) を分離
• DraftPost を公開・承認する (approve) ときには
  • DraftPost に対応する (pk:title が一致する) PublishedPost があればそれを更新し、なければ作成する
• DraftPost を get する際には、対応する PublishedPost が存在するか (has_published_post) をレスポンスに含める
• (ModelViewSet の検証のためだけにぱっと作ったプロジェクトなので、ブログシステムを開発する際にこのような DB 設計にするのが妥当であるかとかは全く考えてないです。)

Model

class PublishedPost(models.Model):
    title = models.CharField(max_length=200, primary_key=True)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    category = models.ForeignKey(
        Category, on_delete=models.SET_NULL, null=True, blank=True
    )
    tags = models.ManyToManyField(Tag, blank=True)
    version = models.PositiveIntegerField(default=1)

    def __str__(self):
        return self.title


class DraftPost(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    category = models.ForeignKey(
        Category, on_delete=models.SET_NULL, null=True, blank=True
    )
    tags = models.ManyToManyField(Tag, blank=True)

    def __str__(self):
        return self.title

Serializer

class DraftPostSerializer(serializers.ModelSerializer):
    has_published_post = serializers.BooleanField(read_only=True)
    tags = TagSerializer(many=True, read_only=True)

    class Meta:
        model = DraftPost
        fields = "__all__"

View

class DraftPostViewSet(viewsets.ModelViewSet):
    queryset = DraftPost.objects.annotate(
        has_published_post=Exists(PublishedPost.objects.filter(title=OuterRef("title")))
    ).prefetch_related("tags")
    serializer_class = DraftPostSerializer

    @action(detail=True, methods=["post"])
    def approve(self, request, pk=None):
        draft_post = self.get_object()
        published_post, created = PublishedPost.objects.update_or_create(
            title=draft_post.title,
            defaults={
                "content": draft_post.content,
                "author": draft_post.author,
                "category": draft_post.category,
                # Increment the version number by 1 to reflect the new version of the published post
                "version": F("version") + 1,
            },
        )
        published_post.tags.set(draft_post.tags.all())
        return Response({"status": "approved"}, status=status.HTTP_200_OK)

以下からためせます

h-amemi/drf_sample_annotate_viewset