diff --git a/openedx_learning/__init__.py b/openedx_learning/__init__.py index 07e30072..3aa9b8ce 100644 --- a/openedx_learning/__init__.py +++ b/openedx_learning/__init__.py @@ -1,4 +1,4 @@ """ Open edX Learning ("Learning Core"). """ -__version__ = "0.15.0" +__version__ = "0.16.0" diff --git a/openedx_tagging/core/tagging/api.py b/openedx_tagging/core/tagging/api.py index 1ccc1992..c72ce6da 100644 --- a/openedx_tagging/core/tagging/api.py +++ b/openedx_tagging/core/tagging/api.py @@ -484,3 +484,35 @@ def delete_tags_from_taxonomy( """ taxonomy = taxonomy.cast() taxonomy.delete_tags(tags, with_subtags) + + +def copy_tags(source_object_id: str, dest_object_id: str): + """ + Copy all tags from one object to another. + + This keeps all not-copied tags and delete all + previous copied tags of the dest object. + If there are not-copied tags that also are in 'source_object_id', + then they become copied. + """ + source_object_tags = get_object_tags( + source_object_id, + ) + copied_tags = ObjectTag.objects.filter( + object_id=dest_object_id, + is_copied=True, + ) + + with transaction.atomic(): + # Delete all copied tags of destination + copied_tags.delete() + + # Copy an create object_tags in destination + for object_tag in source_object_tags: + ObjectTag.objects.update_or_create( + object_id=dest_object_id, + taxonomy_id=object_tag.taxonomy_id, + tag_id=object_tag.tag_id, + defaults={"is_copied": True}, + # Note: _value and _export_id are set automatically + ) diff --git a/openedx_tagging/core/tagging/migrations/0018_objecttag_is_copied.py b/openedx_tagging/core/tagging/migrations/0018_objecttag_is_copied.py new file mode 100644 index 00000000..add2b6e9 --- /dev/null +++ b/openedx_tagging/core/tagging/migrations/0018_objecttag_is_copied.py @@ -0,0 +1,18 @@ +# Generated by Django 4.2.16 on 2024-10-04 19:21 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('oel_tagging', '0017_alter_tagimporttask_status'), + ] + + operations = [ + migrations.AddField( + model_name='objecttag', + name='is_copied', + field=models.BooleanField(default=False, help_text="True if this object tag has been copied from one object to another using 'copy_tags' api function"), + ), + ] diff --git a/openedx_tagging/core/tagging/models/base.py b/openedx_tagging/core/tagging/models/base.py index 3a377587..6b2cccf9 100644 --- a/openedx_tagging/core/tagging/models/base.py +++ b/openedx_tagging/core/tagging/models/base.py @@ -795,6 +795,13 @@ class ObjectTag(models.Model): "Tag associated with this object tag. Provides the tag's 'value' if set." ), ) + is_copied = models.BooleanField( + default=False, + help_text=_( + "True if this object tag has been copied from one object to another" + " using 'copy_tags' api function" + ), + ) _export_id = case_insensitive_char_field( max_length=255, help_text=_( @@ -981,6 +988,7 @@ def copy(self, object_tag: ObjectTag) -> Self: self.tag = object_tag.tag self.taxonomy = object_tag.taxonomy self.object_id = object_tag.object_id + self.is_copied = object_tag.is_copied self._value = object_tag._value # pylint: disable=protected-access self._export_id = object_tag._export_id # pylint: disable=protected-access return self diff --git a/openedx_tagging/core/tagging/rest_api/v1/serializers.py b/openedx_tagging/core/tagging/rest_api/v1/serializers.py index 7e4d7167..9214453b 100644 --- a/openedx_tagging/core/tagging/rest_api/v1/serializers.py +++ b/openedx_tagging/core/tagging/rest_api/v1/serializers.py @@ -176,6 +176,9 @@ def to_representation(self, instance: list[ObjectTag]) -> dict: """ Convert this list of ObjectTags to the serialized dictionary, grouped by Taxonomy """ + # Allows consumers like edx-platform to override this + ObjectTagViewMinimalSerializer = self.context["view"].minimal_serializer_class + can_tag_object_perm = f"{self.app_label}.can_tag_object" by_object: dict[str, dict[str, Any]] = {} for obj_tag in instance: @@ -194,7 +197,7 @@ def to_representation(self, instance: list[ObjectTag]) -> dict: "export_id": obj_tag.export_id, } taxonomies.append(tax_entry) - tax_entry["tags"].append(ObjectTagMinimalSerializer(obj_tag, context=self.context).data) + tax_entry["tags"].append(ObjectTagViewMinimalSerializer(obj_tag, context=self.context).data) return by_object diff --git a/openedx_tagging/core/tagging/rest_api/v1/views.py b/openedx_tagging/core/tagging/rest_api/v1/views.py index 3562074f..6786282b 100644 --- a/openedx_tagging/core/tagging/rest_api/v1/views.py +++ b/openedx_tagging/core/tagging/rest_api/v1/views.py @@ -37,6 +37,7 @@ from .permissions import ObjectTagObjectPermissions, TaxonomyObjectPermissions, TaxonomyTagsObjectPermissions from .serializers import ( ObjectTagListQueryParamsSerializer, + ObjectTagMinimalSerializer, ObjectTagsByTaxonomySerializer, ObjectTagSerializer, ObjectTagUpdateBodySerializer, @@ -443,7 +444,10 @@ class ObjectTagView( * 405 - Method not allowed """ + # Serializer used in `get_queryset` when getting tags per taxonomy serializer_class = ObjectTagSerializer + # Serializer used in the result in `to_representation` in `ObjectTagsByTaxonomySerializer` + minimal_serializer_class = ObjectTagMinimalSerializer permission_classes = [ObjectTagObjectPermissions] lookup_field = "object_id" diff --git a/tests/openedx_tagging/core/tagging/test_api.py b/tests/openedx_tagging/core/tagging/test_api.py index 32142c77..ce32295f 100644 --- a/tests/openedx_tagging/core/tagging/test_api.py +++ b/tests/openedx_tagging/core/tagging/test_api.py @@ -913,3 +913,136 @@ def test_get_object_tag_counts_deleted_disabled(self) -> None: assert tagging_api.get_object_tag_counts("object_*") == {obj1: 1, obj2: 2} tagging_api.add_tag_to_taxonomy(self.taxonomy, "DPANN", parent_tag_value="Archaea") assert tagging_api.get_object_tag_counts("object_*") == {obj1: 2, obj2: 2} + + def test_copy_tags(self) -> None: + obj1 = "object_id1" + obj2 = "object_id2" + + tags_list = [ + { + "value": "English", + "taxonomy": self.language_taxonomy, + }, + { + "value": "DPANN", + "taxonomy": self.taxonomy, + }, + ] + + for tag_object in tags_list: + tagging_api.tag_object(object_id=obj1, taxonomy=tag_object["taxonomy"], tags=[tag_object["value"]]) + + tagging_api.copy_tags(obj1, obj2) + + object_tags = tagging_api.get_object_tags(obj2) + + assert len(object_tags) == 2 + for index, object_tag in enumerate(object_tags): + object_tag.full_clean() + assert object_tag.value == tags_list[index]["value"] + assert not object_tag.is_deleted + assert object_tag.taxonomy == tags_list[index]["taxonomy"] + assert object_tag.object_id == obj2 + assert object_tag.is_copied is True + + def test_copy_tags_with_non_copied(self) -> None: + obj1 = "object_id1" + obj2 = "object_id2" + + tagging_api.tag_object(object_id=obj1, taxonomy=self.language_taxonomy, tags=["English"]) + + tagging_api.tag_object(object_id=obj2, taxonomy=self.taxonomy, tags=["Chordata"]) + tagging_api.tag_object(object_id=obj2, taxonomy=self.free_text_taxonomy, tags=["has a notochord"]) + + tagging_api.copy_tags(obj1, obj2) + object_tags = tagging_api.get_object_tags(obj2) + + # Tags must be the non-copied and the copied tag. + expected_tags = [ + { + "value": "has a notochord", + "taxonomy": self.free_text_taxonomy, + "copied": False, + }, + { + "value": "English", + "taxonomy": self.language_taxonomy, + "copied": True, + }, + { + "value": "Chordata", + "taxonomy": self.taxonomy, + "copied": False, + }, + ] + assert len(object_tags) == 3 + for index, object_tag in enumerate(object_tags): + assert object_tag.value == expected_tags[index]["value"] + assert not object_tag.is_deleted + assert object_tag.taxonomy == expected_tags[index]["taxonomy"] + assert object_tag.object_id == obj2 + assert object_tag.is_copied == expected_tags[index]["copied"] + + # Delete tags of 'obj1' and add other + tagging_api.delete_object_tags(obj1) + tagging_api.tag_object(object_id=obj1, taxonomy=self.taxonomy, tags=["DPANN"]) + tagging_api.copy_tags(obj1, obj2) + object_tags = tagging_api.get_object_tags(obj2) + + # Tags must be the non-copied and the new copied tag. + # The previous copied tags must be deleted. + expected_tags = [ + { + "value": "has a notochord", + "taxonomy": self.free_text_taxonomy, + "copied": False, + }, + { + "value": "DPANN", + "taxonomy": self.taxonomy, + "copied": True, + }, + { + "value": "Chordata", + "taxonomy": self.taxonomy, + "copied": False, + }, + ] + assert len(object_tags) == 3 + for index, object_tag in enumerate(object_tags): + assert object_tag.value == expected_tags[index]["value"] + assert not object_tag.is_deleted + assert object_tag.taxonomy == expected_tags[index]["taxonomy"] + assert object_tag.object_id == obj2 + assert object_tag.is_copied == expected_tags[index]["copied"] + + # Add a tag used by 'obj2' + tagging_api.tag_object(object_id=obj1, taxonomy=self.free_text_taxonomy, tags=["has a notochord"]) + tagging_api.copy_tags(obj1, obj2) + object_tags = tagging_api.get_object_tags(obj2) + + # The non-copied tag 'has a notochord' must be copied now. + expected_tags = [ + { + "value": "has a notochord", + "taxonomy": self.free_text_taxonomy, + "copied": True, + }, + { + "value": "DPANN", + "taxonomy": self.taxonomy, + "copied": True, + }, + { + "value": "Chordata", + "taxonomy": self.taxonomy, + "copied": False, + }, + ] + assert len(object_tags) == 3 + for index, object_tag in enumerate(object_tags): + assert object_tag.value == expected_tags[index]["value"] + assert not object_tag.is_deleted + assert object_tag.taxonomy == expected_tags[index]["taxonomy"] + assert object_tag.object_id == obj2 + assert object_tag.is_copied == expected_tags[index]["copied"]