feat: export tagged course as csv

openedx/core/djangoapps/content_tagging/api.py

@@ -3,12 +3,16 @@
 """
 from __future__ import annotations
 
+from itertools import groupby
+
 import openedx_tagging.core.tagging.api as oel_tagging
-from django.db.models import QuerySet, Exists, OuterRef
-from openedx_tagging.core.tagging.models import Taxonomy
+from django.db.models import Exists, OuterRef, Q, QuerySet
+from opaque_keys.edx.keys import CourseKey, LearningContextKey
+from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
 from organizations.models import Organization
 
 from .models import TaxonomyOrg
+from .types import ObjectTagByObjectIdDict, TaxonomyDict
 
 
 def create_taxonomy(
@@ -126,6 +130,43 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:
     )
 
 
+def get_all_object_tags(
+    content_key: LearningContextKey,
+) -> tuple[ObjectTagByObjectIdDict, TaxonomyDict]:
+    """
+    Returns a tuple with a dictionary of grouped object tags for all blocks and a dictionary of taxonomies.
+    """
+    # ToDo: Add support for other content types (like LibraryContent and LibraryBlock)
+    if isinstance(content_key, CourseKey):
+        course_key_str = str(content_key)
+        # 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.
+        block_id_prefix = course_key_str.replace("course-v1:", "block-v1:", 1)
+    else:
+        raise NotImplementedError(f"Invalid content_key: {type(content_key)} -> {content_key}")
+
+    # There is no API method in oel_tagging.api that does this yet,
+    # so for now we have to build the ORM query directly.
+    all_object_tags = list(ObjectTag.objects.filter(
+        Q(object_id__startswith=block_id_prefix) | Q(object_id=course_key_str),
+        Q(tag__isnull=False, tag__taxonomy__isnull=False),
+    ).select_related("tag__taxonomy"))
+
+    grouped_object_tags: ObjectTagByObjectIdDict = {}
+    taxonomies: TaxonomyDict = {}
+
+    for object_id, block_tags in groupby(all_object_tags, lambda x: x.object_id):
+        grouped_object_tags[object_id] = {}
+        for taxonomy_id, taxonomy_tags in groupby(block_tags, lambda x: x.tag.taxonomy_id):
+            object_tags_list = list(taxonomy_tags)
+            grouped_object_tags[object_id][taxonomy_id] = object_tags_list
+
+            if taxonomy_id not in taxonomies:
+                taxonomies[taxonomy_id] = object_tags_list[0].tag.taxonomy
+
+    return grouped_object_tags, taxonomies
+
+
 # Expose the oel_tagging APIs
 
 get_taxonomy = oel_tagging.get_taxonomy

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

@@ -0,0 +1,88 @@
+"""
+This module contains helper functions to build a object tree with object tags.
+"""
+
+from __future__ import annotations
+
+from typing import Iterator
+
+from attrs import define
+from opaque_keys.edx.keys import CourseKey, LearningContextKey
+
+from xmodule.modulestore.django import modulestore
+
+from ...types import ObjectTagByObjectIdDict, ObjectTagByTaxonomyIdDict
+
+
+@define
+class TaggedContent:
+    """
+    A tagged content, with its tags and children.
+    """
+    display_name: str
+    block_id: str
+    category: str
+    object_tags: ObjectTagByTaxonomyIdDict
+    children: list[TaggedContent] | None
+
+
+def iterate_with_level(
+    tagged_content: TaggedContent, level: int = 0
+) -> Iterator[tuple[TaggedContent, int]]:
+    """
+    Iterator that yields the tagged content and the level of the block
+    """
+    yield tagged_content, level
+    if tagged_content.children:
+        for child in tagged_content.children:
+            yield from iterate_with_level(child, level + 1)
+
+
+def build_object_tree_with_objecttags(
+    content_key: LearningContextKey,
+    object_tag_cache: ObjectTagByObjectIdDict,
+) -> TaggedContent:
+    """
+    Returns the object with the tags associated with it.
+    """
+    store = modulestore()
+
+    if isinstance(content_key, CourseKey):
+        course = store.get_course(content_key)
+        if course is None:
+            raise ValueError(f"Course not found: {content_key}")
+    else:
+        raise NotImplementedError(f"Invalid content_key: {type(content_key)} -> {content_key}")
+
+    display_name = course.display_name_with_default
+    course_id = str(course.id)
+
+    tagged_course = TaggedContent(
+        display_name=display_name,
+        block_id=course_id,
+        category=course.category,
+        object_tags=object_tag_cache.get(str(content_key), {}),
+        children=None,
+    )
+
+    blocks = [(tagged_course, course)]
+
+    while blocks:
+        tagged_block, xblock = blocks.pop()
+        tagged_block.children = []
+
+        if xblock.has_children:
+            for child_id in xblock.children:
+                child_block = store.get_item(child_id)
+                tagged_child = TaggedContent(
+                    display_name=child_block.display_name_with_default,
+                    block_id=str(child_id),
+                    category=child_block.category,
+                    object_tags=object_tag_cache.get(str(child_id), {}),
+                    children=None,
+                )
+                tagged_block.children.append(tagged_child)
+
+                blocks.append((tagged_child, child_block))
+
+    return tagged_course

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

@@ -0,0 +1,177 @@
+"""
+Test the objecttag_export_helpers module
+"""
+from unittest.mock import patch
+
+from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
+
+from .... import api
+from ....tests.test_api import TestGetAllObjectTagsMixin
+from ..objecttag_export_helpers import TaggedContent, build_object_tree_with_objecttags, iterate_with_level
+
+
+class TaggedCourseMixin(TestGetAllObjectTagsMixin, ModuleStoreTestCase):  # type: ignore[misc]
+    """
+    Mixin with a course structure and taxonomies
+    """
+    MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
+    CREATE_USER = False
+
+    def setUp(self):
+        super().setUp()
+
+        # Patch modulestore
+        self.patcher = patch("openedx.core.djangoapps.content_tagging.tasks.modulestore", return_value=self.store)
+        self.addCleanup(self.patcher.stop)
+        self.patcher.start()
+
+        # Create course
+        self.course = CourseFactory.create(
+            org=self.orgA.short_name,
+            number="test_course",
+            run="test_run",
+            display_name="Test Course",
+        )
+        self.expected_tagged_xblock = TaggedContent(
+            display_name="Test Course",
+            block_id="course-v1:orgA+test_course+test_run",
+            category="course",
+            children=[],
+            object_tags={
+                self.taxonomy_1.id: list(self.course_tags),
+            },
+        )
+
+        # Create XBlocks
+        self.sequential = BlockFactory.create(
+            parent=self.course,
+            category="sequential",
+            display_name="test sequential",
+        )
+        # Tag blocks
+        tagged_sequential = TaggedContent(
+            display_name="test sequential",
+            block_id="block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            category="sequential",
+            children=[],
+            object_tags={
+                self.taxonomy_1.id: list(self.sequential_tags1),
+                self.taxonomy_2.id: list(self.sequential_tags2),
+            },
+        )
+
+        assert self.expected_tagged_xblock.children is not None  # type guard
+        self.expected_tagged_xblock.children.append(tagged_sequential)
+
+        # Untagged blocks
+        sequential2 = BlockFactory.create(
+            parent=self.course,
+            category="sequential",
+            display_name="untagged sequential",
+        )
+        untagged_sequential = TaggedContent(
+            display_name="untagged sequential",
+            block_id="block-v1:orgA+test_course+test_run+type@sequential+block@untagged_sequential",
+            category="sequential",
+            children=[],
+            object_tags={},
+        )
+        assert self.expected_tagged_xblock.children is not None  # type guard
+        self.expected_tagged_xblock.children.append(untagged_sequential)
+        BlockFactory.create(
+            parent=sequential2,
+            category="vertical",
+            display_name="untagged vertical",
+        )
+        untagged_vertical = TaggedContent(
+            display_name="untagged vertical",
+            block_id="block-v1:orgA+test_course+test_run+type@vertical+block@untagged_vertical",
+            category="vertical",
+            children=[],
+            object_tags={},
+        )
+        assert untagged_sequential.children is not None  # type guard
+        untagged_sequential.children.append(untagged_vertical)
+        # /Untagged blocks
+
+        vertical = BlockFactory.create(
+            parent=self.sequential,
+            category="vertical",
+            display_name="test vertical1",
+        )
+        tagged_vertical = TaggedContent(
+            display_name="test vertical1",
+            block_id="block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1",
+            category="vertical",
+            children=[],
+            object_tags={
+                self.taxonomy_2.id: list(self.vertical1_tags),
+            },
+        )
+        assert tagged_sequential.children is not None  # type guard
+        tagged_sequential.children.append(tagged_vertical)
+
+        vertical2 = BlockFactory.create(
+            parent=self.sequential,
+            category="vertical",
+            display_name="test vertical2",
+        )
+        untagged_vertical2 = TaggedContent(
+            display_name="test vertical2",
+            block_id="block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical2",
+            category="vertical",
+            children=[],
+            object_tags={},
+        )
+        assert tagged_sequential.children is not None  # type guard
+        tagged_sequential.children.append(untagged_vertical2)
+
+        html = BlockFactory.create(
+            parent=vertical2,
+            category="html",
+            display_name="test html",
+        )
+        tagged_text = TaggedContent(
+            display_name="test html",
+            block_id="block-v1:orgA+test_course+test_run+type@html+block@test_html",
+            category="html",
+            children=[],
+            object_tags={
+                self.taxonomy_2.id: list(self.html_tags),
+            },
+        )
+        assert untagged_vertical2.children is not None  # type guard
+        untagged_vertical2.children.append(tagged_text)
+
+        self.all_object_tags, _ = api.get_all_object_tags(self.course.id)
+        self.expected_tagged_content_list = [
+            (self.expected_tagged_xblock, 0),
+            (tagged_sequential, 1),
+            (tagged_vertical, 2),
+            (untagged_vertical2, 2),
+            (tagged_text, 3),
+            (untagged_sequential, 1),
+            (untagged_vertical, 2),
+        ]
+
+
+class TestContentTagChildrenExport(TaggedCourseMixin):  # type: ignore[misc]
+    """
+    Test helper functions for exporting tagged content
+    """
+    def test_build_object_tree(self) -> None:
+        """
+        Test if we can export a course
+        """
+        with self.assertNumQueries(3):
+            tagged_xblock = build_object_tree_with_objecttags(self.course.id, self.all_object_tags)
+
+        assert tagged_xblock == self.expected_tagged_xblock
+
+    def test_iterate_with_level(self) -> None:
+        """
+        Test if we can iterate over the tagged content in the correct order
+        """
+        tagged_content_list = list(iterate_with_level(self.expected_tagged_xblock))
+        assert tagged_content_list == self.expected_tagged_content_list