feat: export tagged course as csv

openedx/core/djangoapps/content_tagging/api.py

@@ -3,13 +3,22 @@
 """
 from __future__ import annotations
 
+from itertools import groupby
+
 import openedx_tagging.core.tagging.api as oel_tagging
 from django.db.models import Q, QuerySet, Exists, OuterRef
-from openedx_tagging.core.tagging.models import Taxonomy
+from opaque_keys.edx.keys import CourseKey, UsageKey
+from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
 from organizations.models import Organization
+from xmodule.modulestore.django import modulestore
 
 from .models import ContentObjectTag, TaxonomyOrg
-from .types import ContentKey
+from .types import (
+    ContentKey,
+    ObjectTagByObjectIdDict,
+    TaggedContent,
+    TaxonomyDict,
+)
 
 
 def create_taxonomy(
@@ -128,24 +137,30 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:
 def get_content_tags(
     object_key: ContentKey,
     taxonomy_id: int | None = None,
-) -> QuerySet:
+) -> QuerySet[ContentObjectTag]:
     """
     Generates a list of content tags for a given object.
 
     Pass taxonomy to limit the returned object_tags to a specific taxonomy.
     """
-    return oel_tagging.get_object_tags(
+
+    tags = oel_tagging.get_object_tags(
         object_id=str(object_key),
         taxonomy_id=taxonomy_id,
         object_tag_class=ContentObjectTag,
     )
 
+    # Add a generic type to get_object_tags to fix this
+    return tags  # type: ignore
 
+
+# FixMe: The following method (tag_content_object) is only used in tasks.py for auto-tagging. To tag object we are
+# using oel_tagging.tag_object and checking permissions via rule overrides.
 def tag_content_object(
     object_key: ContentKey,
     taxonomy: Taxonomy,
     tags: list,
-) -> QuerySet:
+) -> QuerySet[ContentObjectTag]:
     """
     This is the main API to use when you want to add/update/delete tags from a content object (e.g. an XBlock or
     course).
@@ -175,6 +190,112 @@ def tag_content_object(
     return get_content_tags(object_key, taxonomy_id=taxonomy.id)
 
 
+def get_content_tags_for_object(
+    content_key: ContentKey,
+    include_children: bool,
+) -> tuple[TaggedContent, TaxonomyDict]:
+    """
+    Returns the object with the tags associated with it. If include_children is True, then it will also include
+    the children of the object and their tags.
+    """
+
+    def _get_object_tags(content_key: ContentKey, include_children: bool) -> QuerySet[ObjectTag]:
+        """
+        Return the tags for the object and its children using a single db query.
+        """
+        content_key_str = str(content_key)
+        if not include_children:
+            return ObjectTag.objects.filter(object_id=content_key_str).select_related("tag__taxonomy").all()
+
+        # We use a block_id_prefix (i.e. the modified course id) to get the tags for the children of the Content
+        # (course) in a single db query.
+        # ToDo: Add support for other content types (like LibraryContent and LibraryBlock)
+        if isinstance(content_key, UsageKey):
+            course_key_str = str(content_key.course_key)
+            block_id_prefix = course_key_str.replace("course-v1:", "block-v1:", 1)
+        elif isinstance(content_key, CourseKey):
+            course_key_str = str(content_key)
+            block_id_prefix = str(content_key).replace("course-v1:", "block-v1:", 1)
+        else:
+            raise NotImplementedError(f"Invalid content_key: {type(content_key)} -> {content_key}")
+
+        return ObjectTag.objects.filter(Q(object_id__startswith=block_id_prefix) | Q(object_id=course_key_str)) \
+            .select_related("tag__taxonomy").all()
+
+    def _group_object_tags_by_objectid_taxonomy(
+        all_object_tags: list[ObjectTag]
+    ) -> tuple[ObjectTagByObjectIdDict, TaxonomyDict]:
+        """
+        Returns a tuple with a dictionary of grouped object tags for all blocks and a dictionary of taxonomies.
+        """
+
+        groupedObjectTags: ObjectTagByObjectIdDict = {}
+        taxonomies: TaxonomyDict = {}
+
+        for object_id, block_tags in groupby(all_object_tags, lambda x: x.object_id):
+            groupedObjectTags[object_id] = {}
+            for taxonomy_id, taxonomy_tags in groupby(block_tags, lambda x: x.tag.taxonomy_id):
+                object_tags_list = list(taxonomy_tags)
+                groupedObjectTags[object_id][taxonomy_id] = object_tags_list
+
+                if taxonomy_id not in taxonomies:
+                    taxonomies[taxonomy_id] = object_tags_list[0].tag.taxonomy
+
+        return groupedObjectTags, taxonomies
+
+    def _get_object_with_tags(
+        content_key: ContentKey,
+        include_children: bool,
+        objectTagCache: ObjectTagByObjectIdDict,
+        store
+    ) -> TaggedContent:
+        """
+        Returns the object with the tags associated with it. If include_children is True, then it will also include
+        the children of the object and their tags.
+        """
+        if isinstance(content_key, CourseKey):
+            xblock = store.get_course(content_key)
+        elif isinstance(content_key, UsageKey):
+            xblock = store.get_item(content_key)
+        else:
+            raise NotImplementedError(f"Invalid content_key: {type(content_key)} -> {content_key}")
+
+        tagged_xblock = TaggedContent(
+            xblock=xblock,
+            object_tags=objectTagCache.get(str(content_key), {}),
+            children=None,
+        )
+
+        if not include_children:
+            return tagged_xblock
+
+        blocks = [tagged_xblock]
+
+        while blocks:
+            block = blocks.pop()
+            block.children = []
+
+            if block.xblock.has_children:
+                for child_id in block.xblock.children:
+                    child = TaggedContent(
+                        xblock=store.get_item(child_id),
+                        object_tags=objectTagCache.get(str(child_id), {}),
+                        children=None,
+                    )
+                    block.children.append(child)
+
+                    blocks.append(child)
+
+        return tagged_xblock
+
+    store = modulestore()
+
+    all_blocks_tag_records = list(_get_object_tags(content_key, include_children))
+    objectTagCache, taxonomies = _group_object_tags_by_objectid_taxonomy(all_blocks_tag_records)
+
+    return _get_object_with_tags(content_key, include_children, objectTagCache, store), taxonomies
+
+
 # Expose the oel_tagging APIs
 
 get_taxonomy = oel_tagging.get_taxonomy

openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py

@@ -98,4 +98,10 @@ def get_all_orgs(self, obj) -> bool:
     class Meta:
         model = TaxonomySerializer.Meta.model
         fields = TaxonomySerializer.Meta.fields + ["orgs", "all_orgs"]
-        read_only_fields = ["orgs", "all_orgs"]
+
+
+class ExportContentTagsQueryParamsSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """
+    Serializer for the query params for the export objecttags GET view
+    """
+    include_children = serializers.BooleanField(required=False, default=True)

openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py

@@ -4,8 +4,9 @@
 
 from __future__ import annotations
 
-from urllib.parse import parse_qs, urlparse
 import json
+from io import BytesIO
+from urllib.parse import parse_qs, urlparse
 from unittest.mock import MagicMock
 
 import abc
@@ -39,12 +40,15 @@
 from openedx.core.djangolib.testing.utils import skip_unless_cms
 from openedx.core.lib import blockstore_api
 
+from ....tests.test_api import TaggedCourseMixin
+
 User = get_user_model()
 
 TAXONOMY_ORG_LIST_URL = "/api/content_tagging/v1/taxonomies/"
 TAXONOMY_ORG_DETAIL_URL = "/api/content_tagging/v1/taxonomies/{pk}/"
 TAXONOMY_ORG_UPDATE_ORG_URL = "/api/content_tagging/v1/taxonomies/{pk}/orgs/"
 OBJECT_TAG_UPDATE_URL = "/api/content_tagging/v1/object_tags/{object_id}/?taxonomy={taxonomy_id}"
+OBJECT_TAGS_EXPORT_URL = "/api/content_tagging/v1/object_tags/{object_id}/export/"
 OBJECT_TAGS_URL = "/api/content_tagging/v1/object_tags/{object_id}/"
 TAXONOMY_TEMPLATE_URL = "/api/content_tagging/v1/taxonomies/import/{filename}"
 TAXONOMY_CREATE_IMPORT_URL = "/api/content_tagging/v1/taxonomies/import/"
@@ -1624,6 +1628,91 @@ def test_object_tags_query_count(self):
         assert response.data[object_id]["taxonomies"][0]["tags"] == expected_tags
 
 
+@skip_unless_cms
+@ddt.ddt
+class TestContentObjectChildrenExportView(TaggedCourseMixin, APITestCase):  # type: ignore[misc]
+    """
+    Tests exporting course children with tags
+    """
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.create(
+            username="user",
+            email="[email protected]",
+        )
+        self.staff = User.objects.create(
+            username="staff",
+            email="[email protected]",
+            is_staff=True,
+        )
+
+        self.staffA = User.objects.create(
+            username="staffA",
+            email="[email protected]",
+        )
+        update_org_role(self.staff, OrgStaffRole, self.staffA, [self.orgA.short_name])
+
+    @ddt.data(
+        "staff",
+        "staffA",
+    )
+    def test_export_course(self, user_attr) -> None:
+        url = OBJECT_TAGS_EXPORT_URL.format(object_id=str(self.course.id))
+
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+        response = self.client.get(url)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.headers['Content-Type'] == 'text/csv'
+
+        expected_csv = (
+            "Name,Type,ID,Taxonomy 1,Taxonomy 2\r\n"
+            'Test Course,course,course-v1:orgA+test_course+test_run,Tag 1.1,\r\n'
+            '  test sequential,sequential,block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential,'
+            '"Tag 1.1, Tag 1.2",Tag 2.1\r\n'
+            '    test vertical1,vertical,block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1,'
+            ',Tag 2.2\r\n'
+            '    test vertical2,vertical,block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical2,,\r\n'
+            '      Text,html,block-v1:orgA+test_course+test_run+type@html+block@test_html,,Tag 2.1\r\n'
+        )
+
+        zip_content = BytesIO(b"".join(response.streaming_content)).getvalue()  # type: ignore[attr-defined]
+        assert zip_content == expected_csv.encode()
+
+    def test_export_course_no_children(self) -> None:
+        url = OBJECT_TAGS_EXPORT_URL.format(object_id=str(self.course.id))
+
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.get(url, {"include_children": False})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.headers['Content-Type'] == 'text/csv'
+
+        expected_csv = (
+            "Name,Type,ID,Taxonomy 1\r\n"
+            'Test Course,course,course-v1:orgA+test_course+test_run,Tag 1.1\r\n'
+        )
+
+        zip_content = BytesIO(b"".join(response.streaming_content)).getvalue()  # type: ignore[attr-defined]
+        assert zip_content == expected_csv.encode()
+
+    def test_export_course_anoymous_forbidden(self) -> None:
+        url = OBJECT_TAGS_EXPORT_URL.format(object_id=str(self.course.id))
+        response = self.client.get(url)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_export_course_user_forbidden(self) -> None:
+        url = OBJECT_TAGS_EXPORT_URL.format(object_id=str(self.course.id))
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(url)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_export_course_invalid_id(self) -> None:
+        url = OBJECT_TAGS_EXPORT_URL.format(object_id="invalid")
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.get(url)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
 @skip_unless_cms
 @ddt.ddt
 class TestDownloadTemplateView(APITestCase):
@@ -1635,20 +1724,20 @@ class TestDownloadTemplateView(APITestCase):
         ("template.json", "application/json"),
     )
     @ddt.unpack
-    def test_download(self, filename, content_type):
+    def test_download(self, filename, content_type) -> None:
         url = TAXONOMY_TEMPLATE_URL.format(filename=filename)
         response = self.client.get(url)
         assert response.status_code == status.HTTP_200_OK
         assert response.headers['Content-Type'] == content_type
         assert response.headers['Content-Disposition'] == f'attachment; filename="{filename}"'
         assert int(response.headers['Content-Length']) > 0
 
-    def test_download_not_found(self):
+    def test_download_not_found(self) -> None:
         url = TAXONOMY_TEMPLATE_URL.format(filename="template.txt")
         response = self.client.get(url)
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
-    def test_download_method_not_allowed(self):
+    def test_download_method_not_allowed(self) -> None:
         url = TAXONOMY_TEMPLATE_URL.format(filename="template.txt")
         response = self.client.post(url)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED

openedx/core/djangoapps/content_tagging/rest_api/v1/urls.py

@@ -30,5 +30,9 @@
         oel_tagging_views_import.TemplateView.as_view(),
         name="taxonomy-import-template",
     ),
+    path(
+        "object_tags/<str:object_id>/export/",
+        views.ObjectTagExportView.as_view(),
+    ),
     path('', include(router.urls))
 ]