Skip to content

Commit

Permalink
feat: export tagged content library as CSV (#34246)
Browse files Browse the repository at this point in the history
  • Loading branch information
rpenido authored Mar 4, 2024
1 parent 8bb2f31 commit 42418fb
Show file tree
Hide file tree
Showing 5 changed files with 348 additions and 74 deletions.
21 changes: 13 additions & 8 deletions openedx/core/djangoapps/content_tagging/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -130,24 +131,28 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:


def get_all_object_tags(
content_key: LearningContextKey,
content_key: LibraryLocatorV2 | CourseKey,
) -> tuple[ObjectTagByObjectIdDict, TaxonomyDict]:
"""
Get all the object tags applied to components in the given course/library.
Includes any tags applied to the course/library as a whole.
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"))

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -38,51 +42,137 @@ 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,
)

library_components = library_api.get_library_components(library_key)
children = [
LibraryXBlockMetadata.from_component(library_key, component)
for component in library_components
]

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
Loading

0 comments on commit 42418fb

Please sign in to comment.