feat: export tagged content library as csv [FC-0049]

openedx/core/djangoapps/content_tagging/api.py

@@ -7,7 +7,8 @@
 
 import openedx_tagging.core.tagging.api as oel_tagging
 from django.db.models import Exists, OuterRef, Q, QuerySet
-from opaque_keys.edx.keys import CourseKey, LearningContextKey
+from opaque_keys.edx.keys import CourseKey
+from opaque_keys.edx.locator import LibraryLocatorV2
 from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
 from organizations.models import Organization
 
@@ -130,24 +131,25 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:
 
 
 def get_all_object_tags(
-    content_key: LearningContextKey,
+    content_key: LibraryLocatorV2 | CourseKey,
 ) -> 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)
+    context_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/library) in a single db query.
     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)
+        block_id_prefix = context_key_str.replace("course-v1:", "block-v1:", 1)
+    elif isinstance(content_key, LibraryLocatorV2):
+        block_id_prefix = context_key_str.replace("lib:", "lb:", 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(object_id__startswith=block_id_prefix) | Q(object_id=content_key),
         Q(tag__isnull=False, tag__taxonomy__isnull=False),
     ).select_related("tag__taxonomy"))
 

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

@@ -4,11 +4,15 @@
 
 from __future__ import annotations
 
-from typing import Iterator
+from typing import Any, Callable, Iterator, Union
 
 from attrs import define
-from opaque_keys.edx.keys import CourseKey, LearningContextKey
+from opaque_keys.edx.keys import CourseKey, UsageKey
+from opaque_keys.edx.locator import LibraryLocatorV2
+from xblock.core import XBlock
 
+import openedx.core.djangoapps.content_libraries.api as library_api
+from openedx.core.djangoapps.content_libraries.api import LibraryXBlockMetadata
 from xmodule.modulestore.django import modulestore
 
 from ...types import ObjectTagByObjectIdDict, ObjectTagByTaxonomyIdDict
@@ -38,51 +42,133 @@ def iterate_with_level(
             yield from iterate_with_level(child, level + 1)
 
 
-def build_object_tree_with_objecttags(
-    content_key: LearningContextKey,
-    object_tag_cache: ObjectTagByObjectIdDict,
-) -> TaggedContent:
+def _get_course_tagged_object_and_children(
+    course_key: CourseKey, object_tag_cache: ObjectTagByObjectIdDict
+) -> tuple[TaggedContent, list[XBlock]]:
     """
-    Returns the object with the tags associated with it.
+    Returns a TaggedContent with course metadata with its tags, and its children.
     """
     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}")
+    course = store.get_course(course_key)
+    if course is None:
+        raise ValueError(f"Course not found: {course_key}")
 
-    display_name = course.display_name_with_default
-    course_id = str(course.id)
+    course_id = str(course_key)
 
     tagged_course = TaggedContent(
-        display_name=display_name,
+        display_name=course.display_name_with_default,
         block_id=course_id,
         category=course.category,
-        object_tags=object_tag_cache.get(str(content_key), {}),
+        object_tags=object_tag_cache.get(course_id, {}),
+        children=None,
+    )
+
+    return tagged_course, course.children if course.has_children else []
+
+
+def _get_library_tagged_object_and_children(
+    library_key: LibraryLocatorV2, object_tag_cache: ObjectTagByObjectIdDict
+) -> tuple[TaggedContent, list[LibraryXBlockMetadata]]:
+    """
+    Returns a TaggedContent with library metadata with its tags, and its children.
+    """
+    library = library_api.get_library(library_key)
+    if library is None:
+        raise ValueError(f"Library not found: {library_key}")
+
+    library_id = str(library_key)
+
+    tagged_library = TaggedContent(
+        display_name=library.title,
+        block_id=library_id,
+        category='library',
+        object_tags=object_tag_cache.get(library_id, {}),
+        children=None,
+    )
+
+    children = library_api.get_library_blocks(library_key)
+
+    return tagged_library, children
+
+
+def _get_xblock_tagged_object_and_children(
+    usage_key: UsageKey, object_tag_cache: ObjectTagByObjectIdDict
+) -> tuple[TaggedContent, list[XBlock]]:
+    """
+    Returns a TaggedContent with xblock metadata with its tags, and its children.
+    """
+    store = modulestore()
+    block = store.get_item(usage_key)
+    block_id = str(usage_key)
+    tagged_block = TaggedContent(
+        display_name=block.display_name_with_default,
+        block_id=block_id,
+        category=block.category,
+        object_tags=object_tag_cache.get(block_id, {}),
         children=None,
     )
 
-    blocks = [(tagged_course, course)]
+    return tagged_block, block.children if block.has_children else []
+
+
+def _get_library_block_tagged_object(
+    library_block: LibraryXBlockMetadata, object_tag_cache: ObjectTagByObjectIdDict
+) -> tuple[TaggedContent, None]:
+    """
+    Returns a TaggedContent with library content block metadata and its tags,
+    and 'None' as children.
+    """
+    block_id = str(library_block.usage_key)
+    tagged_library_block = TaggedContent(
+        display_name=library_block.display_name,
+        block_id=block_id,
+        category=library_block.usage_key.block_type,
+        object_tags=object_tag_cache.get(block_id, {}),
+        children=None,
+    )
+
+    return tagged_library_block, None
+
+
+def build_object_tree_with_objecttags(
+    content_key: LibraryLocatorV2 | CourseKey,
+    object_tag_cache: ObjectTagByObjectIdDict,
+) -> TaggedContent:
+    """
+    Returns the object with the tags associated with it.
+    """
+    get_tagged_children: Union[
+        # _get_course_tagged_object_and_children type
+        Callable[[LibraryXBlockMetadata, dict[str, dict[int, list[Any]]]], tuple[TaggedContent, None]],
+        # _get_library_block_tagged_object type
+        Callable[[UsageKey, dict[str, dict[int, list[Any]]]], tuple[TaggedContent, list[Any]]]
+    ]
+    if isinstance(content_key, CourseKey):
+        tagged_content, children = _get_course_tagged_object_and_children(
+            content_key, object_tag_cache
+        )
+        get_tagged_children = _get_xblock_tagged_object_and_children
+    elif isinstance(content_key, LibraryLocatorV2):
+        tagged_content, children = _get_library_tagged_object_and_children(
+            content_key, object_tag_cache
+        )
+        get_tagged_children = _get_library_block_tagged_object
+    else:
+        raise ValueError(f"Invalid content_key: {type(content_key)} -> {content_key}")
+
+    blocks: list[tuple[TaggedContent, list | None]] = [(tagged_content, children)]
 
     while blocks:
-        tagged_block, xblock = blocks.pop()
+        tagged_block, block_children = 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
+        if not block_children:
+            continue
+
+        for child in block_children:
+            tagged_child, child_children = get_tagged_children(child, object_tag_cache)
+            tagged_block.children.append(tagged_child)
+            blocks.append((tagged_child, child_children))
+
+    return tagged_content