feat: retrieve object tags along with their lineage, grouped by taxonomy

openedx_tagging/core/tagging/api.py

@@ -12,12 +12,14 @@
 """
 from __future__ import annotations
 
-from django.db import transaction
-from django.db.models import QuerySet
+from django.db import models, transaction
+from django.db.models import F, QuerySet, Value
+from django.db.models.functions import Coalesce, Concat, Lower
 from django.utils.translation import gettext as _
 
 from .data import TagDataQuerySet
 from .models import ObjectTag, Tag, Taxonomy
+from .models.utils import ConcatNull
 
 # Export this as part of the API
 TagDoesNotExist = Tag.DoesNotExist
@@ -165,8 +167,16 @@ def get_object_tags(
     filters = {"taxonomy_id": taxonomy_id} if taxonomy_id else {}
     tags = (
         object_tag_class.objects.filter(object_id=object_id, **filters)
-        .select_related("tag", "taxonomy")
-        .order_by("id")
+        .select_related("taxonomy", "tag", "tag__parent", "tag__parent__parent")
+        .annotate(sort_key=Lower(Concat(
+            ConcatNull(F("tag__parent__parent__parent__value"), Value("\t")),
+            ConcatNull(F("tag__parent__parent__value"), Value("\t")),
+            ConcatNull(F("tag__parent__value"), Value("\t")),
+            Coalesce(F("tag__value"), F("_value")),
+            Value("\t"),
+            output_field=models.CharField(),
+        )))
+        .order_by("sort_key")
     )
     return tags
 

openedx_tagging/core/tagging/models/base.py

@@ -108,18 +108,18 @@ def get_lineage(self) -> Lineage:
             lineage.insert(0, next_ancestor.value)
             next_ancestor = next_ancestor.get_next_ancestor()
         return lineage
-    
+
     def get_next_ancestor(self) -> Tag | None:
         """
         Fetch the parent of this Tag.
-        
+
         While doing so, preload several ancestors at the same time, so we can
         use fewer database queries than the basic approach of iterating through
         parent.parent.parent...
         """
         if self.parent_id is None:
             return None
-        if not Tag.parent.is_cached(self):
+        if not Tag.parent.is_cached(self):  # pylint: disable=no-member
             # Parent is not yet loaded. Retrieve our parent, grandparent, and great-grandparent in one query.
             # This is not actually changing the parent, just loading it and caching it.
             self.parent = Tag.objects.select_related("parent", "parent__parent").get(pk=self.parent_id)

openedx_tagging/core/tagging/rest_api/v1/serializers.py

@@ -3,6 +3,8 @@
 """
 from __future__ import annotations
 
+from typing import Any
+
 from rest_framework import serializers
 from rest_framework.reverse import reverse
 
@@ -61,24 +63,59 @@ class ObjectTagListQueryParamsSerializer(serializers.Serializer):  # pylint: dis
     )
 
 
-class ObjectTagSerializer(serializers.ModelSerializer):
+class ObjectTagMinimalSerializer(serializers.ModelSerializer):
     """
-    Serializer for the ObjectTag model.
+    Minimal serializer for the ObjectTag model.
     """
 
     class Meta:
         model = ObjectTag
-        fields = [
+        fields = ["value", "lineage"]
+
+    lineage = serializers.ListField(child=serializers.CharField(), source="get_lineage")
+
+
+class ObjectTagSerializer(ObjectTagMinimalSerializer):
+    """
+    Serializer for the ObjectTag model.
+    """
+    class Meta:
+        fields = ObjectTagMinimalSerializer.Meta.fields + [
             # The taxonomy name
             "name",
-            "value",
-            "lineage",
             "taxonomy_id",
             # If the Tag or Taxonomy has been deleted, this ObjectTag shouldn't be shown to users.
             "is_deleted",
         ]
 
-    lineage = serializers.ListField(child=serializers.CharField(), source="get_lineage")
+
+class ObjectTagsByTaxonomySerializer(serializers.ModelSerializer):
+    """
+    Serialize a group of ObjectTags, grouped by taxonomy
+    """
+    def to_representation(self, instance: list[ObjectTag]) -> dict:
+        """
+        Convert this list of ObjectTags to the serialized dictionary, grouped by Taxonomy
+        """
+        by_object: dict[str, dict[str, Any]] = {}
+        for obj_tag in instance:
+            if obj_tag.object_id not in by_object:
+                by_object[obj_tag.object_id] = {
+                    "taxonomies": []
+                }
+            taxonomies = by_object[obj_tag.object_id]["taxonomies"]
+            tax_entry = next((t for t in taxonomies if t["taxonomy_id"] == obj_tag.taxonomy_id), None)
+            if tax_entry is None:
+                tax_entry = {
+                    "name": obj_tag.name,
+                    "taxonomy_id": obj_tag.taxonomy_id,
+                    "editable": (not obj_tag.taxonomy.cast().system_defined) if obj_tag.taxonomy else False,
+                    "tags": []
+                }
+                taxonomies.append(tax_entry)
+            tax_entry["tags"].append(ObjectTagMinimalSerializer(obj_tag).data)
+        return by_object
+
 
 class ObjectTagUpdateBodySerializer(serializers.Serializer):  # pylint: disable=abstract-method
     """

openedx_tagging/core/tagging/rest_api/v1/views.py

@@ -32,6 +32,7 @@
 from .permissions import ObjectTagObjectPermissions, TagObjectPermissions, TaxonomyObjectPermissions
 from .serializers import (
     ObjectTagListQueryParamsSerializer,
+    ObjectTagsByTaxonomySerializer,
     ObjectTagSerializer,
     ObjectTagUpdateBodySerializer,
     ObjectTagUpdateQueryParamsSerializer,
@@ -306,20 +307,20 @@ def get_queryset(self) -> models.QuerySet:
             taxonomy = taxonomy.cast()
             taxonomy_id = taxonomy.id
 
-        perm = "oel_tagging.view_objecttag"
-        perm_obj = ObjectTagPermissionItem(
-            taxonomy=taxonomy,
-            object_id=object_id,
-        )
-
-        if not self.request.user.has_perm(
-            perm,
-            # The obj arg expects a model, but we are passing an object
-            perm_obj,  # type: ignore[arg-type]
-        ):
-            raise PermissionDenied(
-                "You do not have permission to view object tags for this taxonomy or object_id."
-            )
+        if object_id.endswith("*") or "," in object_id:  # pylint: disable=no-else-raise
+            raise ValidationError("Retrieving tags from multiple objects is not yet supported.")
+            # Note: This API is actually designed so that in the future it can be extended to return tags for multiple
+            # objects, e.g. if object_id.endswith("*") then it results in a object_id__startswith query. However, for
+            # now we have no use case for that so we retrieve tags for one object at a time.
+        else:
+            if not self.request.user.has_perm(
+                "oel_tagging.view_objecttag",
+                # The obj arg expects a model, but we are passing an object
+                ObjectTagPermissionItem(taxonomy=taxonomy, object_id=object_id),  # type: ignore[arg-type]
+            ):
+                raise PermissionDenied(
+                    "You do not have permission to view object tags for this taxonomy or object_id."
+                )
 
         return get_object_tags(object_id, taxonomy_id)
 
@@ -334,9 +335,14 @@ def retrieve(self, request, *args, **kwargs) -> Response:
         path and returns a it as a single result however that is not
         behavior we want.
         """
-        object_tags = self.filter_queryset(self.get_queryset())
-        serializer = ObjectTagSerializer(object_tags, many=True)
-        return Response(serializer.data)
+        object_tags = self.filter_queryset(self.get_queryset()).order_by("sort_key")
+        serializer = ObjectTagsByTaxonomySerializer(list(object_tags))
+        response_data = serializer.data
+        if self.kwargs["object_id"] not in response_data:
+            # For consistency, the key with the object_id should always be present in the response, even if there
+            # are no tags at all applied to this object.
+            response_data[self.kwargs["object_id"]] = {"taxonomies": []}
+        return Response(response_data)
 
     def update(self, request, *args, **kwargs) -> Response:
         """

tests/openedx_tagging/core/tagging/test_api.py

@@ -85,17 +85,19 @@ def test_get_taxonomies(self) -> None:
             enabled = list(tagging_api.get_taxonomies())
         assert enabled == [
             tax1,
+            self.free_text_taxonomy,
             tax3,
             self.language_taxonomy,
             self.taxonomy,
             self.system_taxonomy,
             self.user_taxonomy,
-        ] + self.dummy_taxonomies
+        ]
         assert str(enabled[0]) == f"<Taxonomy> ({tax1.id}) Enabled"
-        assert str(enabled[1]) == "<Taxonomy> (5) Import Taxonomy Test"
-        assert str(enabled[2]) == "<LanguageTaxonomy> (-1) Languages"
-        assert str(enabled[3]) == "<Taxonomy> (1) Life on Earth"
-        assert str(enabled[4]) == "<SystemDefinedTaxonomy> (4) System defined taxonomy"
+        assert str(enabled[1]) == "<Taxonomy> (6) Free Text"
+        assert str(enabled[2]) == "<Taxonomy> (5) Import Taxonomy Test"
+        assert str(enabled[3]) == "<LanguageTaxonomy> (-1) Languages"
+        assert str(enabled[4]) == "<Taxonomy> (1) Life on Earth"
+        assert str(enabled[5]) == "<SystemDefinedTaxonomy> (4) System defined taxonomy"
 
         with self.assertNumQueries(1):
             disabled = list(tagging_api.get_taxonomies(enabled=False))
@@ -107,12 +109,13 @@ def test_get_taxonomies(self) -> None:
         assert both == [
             tax2,
             tax1,
+            self.free_text_taxonomy,
             tax3,
             self.language_taxonomy,
             self.taxonomy,
             self.system_taxonomy,
             self.user_taxonomy,
-        ] + self.dummy_taxonomies
+        ]
 
     def test_get_tags(self) -> None:
         assert pretty_format_tags(tagging_api.get_tags(self.taxonomy), parent=False) == [
@@ -265,18 +268,18 @@ def test_resync_object_tags(self) -> None:
         assert [(t.value, t.is_deleted) for t in tagging_api.get_object_tags(object_id)] == [
             (self.archaea.value, False),
             (self.bacteria.value, False),
-            ("foo", False),
             ("bar", False),
+            ("foo", False),
         ]
 
         # Delete "bacteria" from the taxonomy:
-        self.bacteria.delete()  # TODO: add an API method for this
+        tagging_api.delete_tags_from_taxonomy(self.taxonomy, [self.bacteria.value], with_subtags=True)
 
         assert [(t.value, t.is_deleted) for t in tagging_api.get_object_tags(object_id)] == [
             (self.archaea.value, False),
             (self.bacteria.value, True),  # <--- deleted! But the value is preserved.
-            ("foo", False),
             ("bar", False),
+            ("foo", False),
         ]
 
         # Re-syncing the tags at this point does nothing:
@@ -293,8 +296,8 @@ def test_resync_object_tags(self) -> None:
         assert [(t.value, t.is_deleted) for t in tagging_api.get_object_tags(object_id)] == [
             (self.archaea.value, False),
             (self.bacteria.value, False),  # <--- not deleted
-            ("foo", False),
             ("bar", False),
+            ("foo", False),
         ]
 
         # Re-syncing the tags now does nothing:
@@ -311,12 +314,12 @@ def test_tag_object(self):
                 self.chordata,
             ],
             [
-                self.chordata,
                 self.archaebacteria,
+                self.chordata,
             ],
             [
-                self.archaebacteria,
                 self.archaea,
+                self.archaebacteria,
             ],
         ]
 
@@ -520,8 +523,9 @@ def test_tag_object_limit(self) -> None:
         """
         Test that the tagging limit is enforced.
         """
+        dummy_taxonomies = self.create_100_taxonomies()
         # The user can add up to 100 tags to a object
-        for taxonomy in self.dummy_taxonomies:
+        for taxonomy in dummy_taxonomies:
             tagging_api.tag_object(
                 taxonomy,
                 ["Dummy Tag"],
@@ -539,15 +543,15 @@ def test_tag_object_limit(self) -> None:
         assert "Cannot add more than 100 tags to" in str(exc.exception)
 
         # Updating existing tags should work
-        for taxonomy in self.dummy_taxonomies:
+        for taxonomy in dummy_taxonomies:
             tagging_api.tag_object(
                 taxonomy,
                 ["New Dummy Tag"],
                 "object_1",
             )
 
         # Updating existing tags adding a new one should fail
-        for taxonomy in self.dummy_taxonomies:
+        for taxonomy in dummy_taxonomies:
             with self.assertRaises(ValueError) as exc:
                 tagging_api.tag_object(
                     taxonomy,
@@ -558,15 +562,16 @@ def test_tag_object_limit(self) -> None:
             assert "Cannot add more than 100 tags to" in str(exc.exception)
 
     def test_get_object_tags(self) -> None:
-        # Alpha tag has no taxonomy
+        # Alpha tag has no taxonomy (as if the taxonomy had been deleted)
         alpha = ObjectTag(object_id="abc")
         alpha.name = self.taxonomy.name
-        alpha.value = self.mammalia.value
+        alpha.value = "alpha"
         alpha.save()
         # Beta tag has a closed taxonomy
         beta = ObjectTag.objects.create(
             object_id="abc",
             taxonomy=self.taxonomy,
+            tag=self.taxonomy.tag_set.get(value="Protista"),
         )
 
         # Fetch all the tags for a given object ID