Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: export all course tags as csv (#34091)

Browse Source
This commit is contained in:
Rômulo Penido
2024-02-16 14:56:30 -03:00
committed by GitHub
parent 95b3e88ba5
commit 4d1d82dd35
8 changed files with 632 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
openedx/core/djangoapps/content_tagging/api.py
View File

@@ -3,12 +3,16 @@ Content Tagging APIs
"""
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

88
openedx/core/djangoapps/content_tagging/rest_api/v1/objecttag_export_helpers.py Normal file
View File

@@ -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

177
openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_objecttag_export_helpers.py Normal file
View File

@@ -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

98
openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py
View File

@@ -4,11 +4,12 @@ Tests tagging rest api views
from __future__ import annotations
from urllib.parse import parse_qs, urlparse
import json
from unittest.mock import MagicMock
import abc
import json
from io import BytesIO
from unittest.mock import MagicMock
from urllib.parse import parse_qs, urlparse
import ddt
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
@@ -27,24 +28,24 @@ from common.djangoapps.student.roles import (
OrgContentCreatorRole,
OrgInstructorRole,
OrgLibraryUserRole,
OrgStaffRole,
)
from openedx.core.djangoapps.content_libraries.api import (
AccessLevel,
create_library,
set_library_user_permissions,
OrgStaffRole
)
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.content_libraries.api import AccessLevel, create_library, set_library_user_permissions
from openedx.core.djangoapps.content_tagging import api as tagging_api
from openedx.core.djangoapps.content_tagging.models import TaxonomyOrg
from openedx.core.djangolib.testing.utils import skip_unless_cms
from openedx.core.lib import blockstore_api
from .test_objecttag_export_helpers 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/"
@@ -1782,6 +1783,77 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
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 = UserFactory.create()
self.staff = UserFactory.create(
username="staff",
email="staff@example.com",
is_staff=True,
)
self.staffA = UserFactory.create(
username="staffA",
email="userA@example.com",
)
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","1-taxonomy-1","2-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'
'" test html","html","block-v1:orgA+test_course+test_run+type@html+block@test_html","","Tag 2.1"\r\n'
'" untagged sequential","sequential","block-v1:orgA+test_course+test_run+type@sequential+block@untagged_'
'sequential","",""\r\n'
'" untagged vertical","vertical","block-v1:orgA+test_course+test_run+type@vertical+block@untagged_'
'vertical","",""\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):
@@ -1793,7 +1865,7 @@ 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
@@ -1801,12 +1873,12 @@ class TestDownloadTemplateView(APITestCase):
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

16
openedx/core/djangoapps/content_tagging/rest_api/v1/urls.py
View File

@@ -2,15 +2,11 @@
Taxonomies API v1 URLs.
"""
from rest_framework.routers import DefaultRouter
from django.urls.conf import include, path
from openedx_tagging.core.tagging.rest_api.v1 import views as oel_tagging_views
from openedx_tagging.core.tagging.rest_api.v1 import views_import as oel_tagging_views_import
from openedx_tagging.core.tagging.rest_api.v1.views import ObjectTagCountsView
from django.urls.conf import path, include
from openedx_tagging.core.tagging.rest_api.v1 import (
views as oel_tagging_views,
views_import as oel_tagging_views_import,
)
from rest_framework.routers import DefaultRouter
from . import views
@@ -30,5 +26,9 @@ urlpatterns = [
oel_tagging_views_import.TemplateView.as_view(),
name="taxonomy-import-template",
),
path(
"object_tags/<str:context_id>/export/",
views.ObjectTagExportView.as_view(),
),
path('', include(router.urls))
]

97
openedx/core/djangoapps/content_tagging/rest_api/v1/views.py
View File

@@ -1,6 +1,14 @@
"""
Tagging Org API Views
"""
from __future__ import annotations
import csv
from typing import Iterator
from django.http import StreamingHttpResponse
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from openedx_tagging.core.tagging import rules as oel_tagging_rules
from openedx_tagging.core.tagging.rest_api.v1.views import ObjectTagView, TaxonomyView
from rest_framework import status
@@ -8,18 +16,21 @@ from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied, ValidationError
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView
from ...api import (
create_taxonomy,
get_taxonomy,
get_all_object_tags,
get_taxonomies,
get_taxonomies_for_org,
get_taxonomy,
get_unassigned_taxonomies,
set_taxonomy_orgs,
set_taxonomy_orgs
)
from ...rules import get_admin_orgs
from .serializers import TaxonomyOrgListQueryParamsSerializer, TaxonomyOrgSerializer, TaxonomyUpdateOrgBodySerializer
from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend
from .objecttag_export_helpers import build_object_tree_with_objecttags, iterate_with_level
from .serializers import TaxonomyOrgListQueryParamsSerializer, TaxonomyOrgSerializer, TaxonomyUpdateOrgBodySerializer
class TaxonomyOrgView(TaxonomyView):
@@ -135,3 +146,83 @@ class ObjectTagOrgView(ObjectTagView):
Refer to ObjectTagView docstring for usage details.
"""
filter_backends = [ObjectTagTaxonomyOrgFilterBackend]
class ObjectTagExportView(APIView):
""""
View to export a CSV with all children and tags for a given course/context.
"""
def get(self, request: Request, **kwargs) -> StreamingHttpResponse:
"""
Export a CSV with all children and tags for a given course/context.
"""
class Echo(object):
"""
Class that implements just the write method of the file-like interface,
used for the streaming response.
"""
def write(self, value):
return value
def _generate_csv_rows() -> Iterator[str]:
"""
Receives the blocks, tags and taxonomies and returns a CSV string
"""
header = {"name": "Name", "type": "Type", "id": "ID"}
# Prepare the header for the taxonomies
for taxonomy_id, taxonomy in taxonomies.items():
header[f"taxonomy_{taxonomy_id}"] = taxonomy.export_id
csv_writer = csv.DictWriter(pseudo_buffer, fieldnames=header.keys(), quoting=csv.QUOTE_NONNUMERIC)
yield csv_writer.writerow(header)
# Iterate over the blocks and yield the rows
for item, level in iterate_with_level(tagged_content):
block_data = {
"name": level * " " + item.display_name,
"type": item.category,
"id": item.block_id,
}
# Add the tags for each taxonomy
for taxonomy_id in taxonomies:
if taxonomy_id in item.object_tags:
block_data[f"taxonomy_{taxonomy_id}"] = ", ".join([
object_tag.value
for object_tag in item.object_tags[taxonomy_id]
])
yield csv_writer.writerow(block_data)
object_id: str = kwargs.get('context_id', None)
try:
content_key = CourseKey.from_string(object_id)
except InvalidKeyError as e:
raise ValidationError("context_id is not a valid course key.") from e
# Check if the user has permission to view object tags for this object_id
try:
if not self.request.user.has_perm(
"oel_tagging.view_objecttag",
# The obj arg expects a model, but we are passing an object
oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=object_id), # type: ignore[arg-type]
):
raise PermissionDenied(
"You do not have permission to view object tags for this object_id."
)
except ValueError as e:
raise ValidationError from e
all_object_tags, taxonomies = get_all_object_tags(content_key)
tagged_content = build_object_tree_with_objecttags(content_key, all_object_tags)
pseudo_buffer = Echo()
return StreamingHttpResponse(
streaming_content=_generate_csv_rows(),
content_type="text/csv",
headers={'Content-Disposition': f'attachment; filename="{object_id}_tags.csv"'},
)

130
openedx/core/djangoapps/content_tagging/tests/test_api.py
View File

@@ -1,7 +1,8 @@
"""Tests for the Tagging models"""
import ddt
from django.test.testcases import TestCase
from openedx_tagging.core.tagging.models import Tag
from opaque_keys.edx.keys import CourseKey
from openedx_tagging.core.tagging.models import ObjectTag, Tag
from organizations.models import Organization
from .. import api
@@ -231,3 +232,130 @@ class TestAPITaxonomy(TestTaxonomyMixin, TestCase):
assert result[0]["_id"] == self.tag_all_orgs.id
assert result[0]["parent_value"] is None
assert result[0]["depth"] == 0
class TestGetAllObjectTagsMixin:
"""
Set up data to test get_all_object_tags functions
"""
def setUp(self):
super().setUp()
self.orgA = Organization.objects.create(name="Organization A", short_name="orgA")
self.taxonomy_1 = api.create_taxonomy(name="Taxonomy 1")
api.set_taxonomy_orgs(self.taxonomy_1, all_orgs=True)
Tag.objects.create(
taxonomy=self.taxonomy_1,
value="Tag 1.1",
)
Tag.objects.create(
taxonomy=self.taxonomy_1,
value="Tag 1.2",
)
self.taxonomy_2 = api.create_taxonomy(name="Taxonomy 2")
api.set_taxonomy_orgs(self.taxonomy_2, all_orgs=True)
Tag.objects.create(
taxonomy=self.taxonomy_2,
value="Tag 2.1",
)
Tag.objects.create(
taxonomy=self.taxonomy_2,
value="Tag 2.2",
)
api.tag_object(
object_id="course-v1:orgA+test_course+test_run",
taxonomy=self.taxonomy_1,
tags=['Tag 1.1'],
)
self.course_tags = api.get_object_tags("course-v1:orgA+test_course+test_run")
# Tag blocks
api.tag_object(
object_id="block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
taxonomy=self.taxonomy_1,
tags=['Tag 1.1', 'Tag 1.2'],
)
self.sequential_tags1 = api.get_object_tags(
"block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
taxonomy_id=self.taxonomy_1.id,
)
api.tag_object(
object_id="block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
taxonomy=self.taxonomy_2,
tags=['Tag 2.1'],
)
self.sequential_tags2 = api.get_object_tags(
"block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
taxonomy_id=self.taxonomy_2.id,
)
api.tag_object(
object_id="block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1",
taxonomy=self.taxonomy_2,
tags=['Tag 2.2'],
)
self.vertical1_tags = api.get_object_tags(
"block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1"
)
api.tag_object(
object_id="block-v1:orgA+test_course+test_run+type@html+block@test_html",
taxonomy=self.taxonomy_2,
tags=['Tag 2.1'],
)
self.html_tags = api.get_object_tags("block-v1:orgA+test_course+test_run+type@html+block@test_html")
# Create "deleted" object tags, which will be omitted from the results.
for object_id in (
"course-v1:orgA+test_course+test_run",
"block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
"block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1",
"block-v1:orgA+test_course+test_run+type@html+block@test_html",
):
ObjectTag.objects.create(
object_id=str(object_id),
taxonomy=None,
tag=None,
_value="deleted tag",
_name="deleted taxonomy",
)
self.expected_objecttags = {
"course-v1:orgA+test_course+test_run": {
self.taxonomy_1.id: list(self.course_tags),
},
"block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential": {
self.taxonomy_1.id: list(self.sequential_tags1),
self.taxonomy_2.id: list(self.sequential_tags2),
},
"block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1": {
self.taxonomy_2.id: list(self.vertical1_tags),
},
"block-v1:orgA+test_course+test_run+type@html+block@test_html": {
self.taxonomy_2.id: list(self.html_tags),
},
}
class TestGetAllObjectTags(TestGetAllObjectTagsMixin, TestCase):
"""
Test get_all_object_tags api function
"""
def test_get_all_object_tags(self):
"""
Test the get_all_object_tags function
"""
with self.assertNumQueries(1):
object_tags, taxonomies = api.get_all_object_tags(
CourseKey.from_string("course-v1:orgA+test_course+test_run")
)
assert object_tags == self.expected_objecttags
assert taxonomies == {
self.taxonomy_1.id: self.taxonomy_1,
self.taxonomy_2.id: self.taxonomy_2,
}

9
openedx/core/djangoapps/content_tagging/types.py
View File

@@ -1,9 +1,16 @@
"""
Types used by content tagging API and implementation
"""
from typing import Union
from __future__ import annotations
from typing import Dict, List, Union
from opaque_keys.edx.keys import CourseKey, UsageKey
from opaque_keys.edx.locator import LibraryLocatorV2
from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
ContentKey = Union[LibraryLocatorV2, CourseKey, UsageKey]
ObjectTagByTaxonomyIdDict = Dict[int, List[ObjectTag]]
ObjectTagByObjectIdDict = Dict[str, ObjectTagByTaxonomyIdDict]
TaxonomyDict = Dict[int, Taxonomy]