Django queryset: annotate with calculated value

I am making a very simple notification system for my website, powered by a Django REST Framework API. It's for sending website updates and things to all users, everyone gets the same notifications, and they can then mark it as read / archive it. I have come up with the following model:

class Notification(models.Model):
    title = models.CharField(max_length=255)
    text = models.TextField()
    type = models.CharField(max_length=255, blank=True)
    read_by = models.ManyToManyField(User, blank=True, related_name="read_notifications")
    archived_by = models.ManyToManyField(User, blank=True, related_name="archived_notifications")
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
    updated_at = models.DateTimeField(auto_now=True)

So there is no receiver field or something like that, as all users get all notifications anyway.

Now I am trying to write the view logic, notably the following 2 things: only fetch non-archived notifications made after the user was created, and add a calculated "is_read" field to it, in a way that doesn't do extra queries for every single notification / user combination.

The query looks like this now:

queryset = Notification.objects

This does indeed filter out archived queries as expected, and I think it's doing it without an extra query for every notification:

SELECT "notifications_notification"."id", "notifications_notification"."title", "notifications_notification"."text", "notifications_notification"."type", "notifications_notification"."created_at", "notifications_notification"."updated_at" FROM "notifications_notification" WHERE ("notifications_notification"."created_at" > 2022-09-26 12:44:04.771961+00:00 AND NOT (EXISTS(SELECT 1 AS "a" FROM "notifications_notification_archived_by" U1 WHERE (U1."user_id" = 1 AND U1."notification_id" = ("notifications_notification"."id")) LIMIT 1))) ORDER BY "notifications_notification"."created_at" DESC

So far so goo! But I still need to add an "is_read" value (or "is_unread" if easier) to the query somehow, which I am not able to work out how to do.

How can I finish the query and make it performant as well?

After trial and error I came up with this:

queryset = Notification.objects.order_by("-created_at")
    .annotate(is_read=Exists(Notification.objects.filter(pk=OuterRef("id"), read_by=self.request.user)))

And that works, although it does do 2 subqueries and I wonder if this is going to become a bottleneck later on?

SELECT "notifications_notification"."id", "notifications_notification"."title", "notifications_notification"."text", "notifications_notification"."type", "notifications_notification"."created_at", "notifications_notification"."updated_at", EXISTS(SELECT 1 AS "a" FROM "notifications_notification" U0 INNER JOIN "notifications_notification_read_by" U1 ON (U0."id" = U1."notification_id") WHERE (U0."id" = ("notifications_notification"."id") AND U1."user_id" = 1) LIMIT 1) AS "is_read" FROM "notifications_notification" WHERE ("notifications_notification"."created_at" > 2022-09-26 14:40:29.368043+00:00 AND NOT (EXISTS(SELECT 1 AS "a" FROM "notifications_notification_archived_by" U1 WHERE (U1."user_id" = 1 AND U1."notification_id" = ("notifications_notification"."id")) LIMIT 1))) ORDER BY "notifications_notification"."created_at" DESC

Back to Top