Lookup using annotation from a subquery

Django Internals ORM
1

Hello everybody,

My Product model can have multiple children Tagging models, the last of which is of special importance:

class Product(Model):
    name = CharField(...)

class Tagging(Model):
    class Meta:
        ordering = ["-created_at"]
    
    product = ForeignKey(Product, ...)
    created_at = DateTimeField(auto_now_add=True)
    is_active = BooleanField()

>>> Product.objects.filter(taggings__is_active=True).count()
31314

I’m using annotation to get the last version of a product’s tagging like this:

>>> last_tagging = Tagging.objects.filter(product=OuterRef("pk")).values("pk")[0:]
>>> queryset = Product.objects.annotate(last_tagging=Subquery(last_tagging))
>>> queryset.first().last_tagging.is_active
True

The property last_tagging created on the Product instance “resolves” the corresponding Tagging instance from its annotated ID.
That’s helpful, however I expectedly can’t use that annotation with a lookup in the queryset:

>>> queryset.filter(last_tagging__is_active=True).count()
FieldError: Unsupported lookup 'is_active' for UUIDField or join on the field not permitted.

How would I make it possible to “resolve” an annotation from a subquery to enable lookups and prefetches? Is there any workaround?

2

Hello there,

Subquery annotations only support a single field as they are not implemented using JOINs.

I would expect this to crash with

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'int' object has no attribute 'is_active'

As last_tagging will be Tagging.pk: int and not an instance of Tagging .

To support your filtering use case you should annotate the is_active value instead

last_tagging = Tagging.objects.filter(
    product=OuterRef("pk")
).values("is_active")[0:]
queryset = Product.objects.annotate(
    last_tagging_is_active=Subquery(last_tagging)
)
queryset.filter(last_tagging_is_active=True).count()
3

Hi Simon, thank you for the super the quick reply

You’re absolutely right, that was it ambiguous, I should mention that I instantiate the related object from that annotated parameter in my model, that is not Django’s standard behaviour.

Apologies again, my use case example was poorly illustrated: I would need to not only filter is_active, but potentially any other fields of that model with the ORM, I would also need to prefetch other tables through last_tagging like so:

queryset.prefetch_related(Prefetch("last_tagging__tags", queryset=Tag.objects.filter(is_public=True)))
4

Apologies again, my use case example was poorly illustrated: I would need to not only filter is_active, but potentially any other fields of that model with the ORM, I would also need to prefetch other tables through last_tagging like so:

There’s a lot to unfold here but maybe this article might be of some help.

Basically you’d want to annotate a JSONObject representation of your tagging and tags but that’ll be hard to make it work with prefetch_related. Give you already have a method that materialize last_tagging you could something like

last_tagging = Tagging.objects.filter(
    product=OuterRef("pk")
).values(
    JSONObject(id="id", created_at="created_at", is_active="is_active")
)[0:]
queryset = Product.objects.annotate(
    last_tagging=Subquery(last_tagging)
)
queryset.filter(last_tagging__is_active=True).count()

class Tagging(models.Models):
    ...

    def _get_last_tagging(self):
        try:
            return self.__last_tagging
        except AttributeError:
            ... # Retrieve from database and store

    def _set_last_tagging(self, value):
        if isinstance(value, dict):
            value = self.__class__(**value)
        self.__last_tagging = value

    last_tagging = property(_get_last_tagging, _set_last_tagging)

from django.db.models import prefetch_related_objects

prefetch_related_objects(
    [product.last_tagging for product in queryset], "tags"
)
5

Thank you, that does wonders for sort of selecting the latest taggings in a single query, filtering and getting related models. Here is what I have so far:

class ProductQuerySet(QuerySet):
    def with_latest_taggings(self):
        """\
        Provides the instances' latest_tagging properties with prefetched info.
        Lookups work with this method.
        Prefetching related models does not work with this method.
        """

        fields = [(_f.attname, _f.attname,) for _f in Tagging._meta.fields]

        latest_tagging = Tagging.objects.filter(
            product_id=OuterRef("pk"),
        ).values(
            latest_tagging=JSONObject(**dict(fields)),
        )[0:]
        
        return self.annotate(
            _latest_tagging=Subquery(latest_tagging),
        )

class Product(Model):
    objects = ProductQuerySet.as_manager()

    name = CharField(...)

    @property
    def latest_tagging(self):
        if _latest_tagging := getattr(self, "_latest_tagging", None):
            return Tagging(**_latest_tagging)
        else:
            return self.taggings.last()

class Tagging(Model):
    class Meta:
        ordering = ["-created_at"]
    
    product = ForeignKey(Product, ...)
    created_at = DateTimeField(auto_now_add=True)
    is_active = BooleanField()
    # bunch of other fields

>>> queryset = Product.objects.with_latest_taggings()
>>> queryset.filter(_latest_tagging__created_at__gt=last_week).first().latest_tagging.is_active
True
>>> queryset.first().latest_tagging.tags.count()
13

As you said, prefetching through JSONObjects doesn’t work, although it doesn’t raise any exception.

# without the queryset method
>>> public_tags = Tag.objects.filter(is_public=True)
>>> Product.objects.first().latest_tagging.tags.count()
13
>>> Product.objects.prefetch_related(Prefetch("taggings__tags", queryset=public_tags)).first().latest_tagging.tags.count()
11
# with the queryset method
>>> queryset.first().latest_tagging.tags.count()
13
>>> queryset.prefetch_related(Prefetch("_latest_tagging__tags", queryset=public_tags)).first().latest_tagging.tags.count()
13

With that limitation I’m a bit afraid it’s not worth the gain.