feat: Copy object tags [FC-0062]

openedx_learning/__init__.py

@@ -1,4 +1,4 @@
 """
 Open edX Learning ("Learning Core").
 """
-__version__ = "0.14.0"
+__version__ = "0.15.0"

openedx_tagging/core/tagging/api.py

@@ -484,3 +484,60 @@ 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,
+    include_deleted: bool = True,
+    object_tag_class: type[ObjectTag] = ObjectTag
+):
+    """
+    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.
+    """
+    ObjectTagClass = object_tag_class
+
+    def dest_object_has_tag(object_tag):
+        try:
+            result = ObjectTagClass.objects.get(
+                object_id=dest_object_id,
+                _export_id=object_tag.export_id,
+                _value=object_tag.value,
+            )
+            return result
+        except ObjectTagClass.DoesNotExist:
+            return None
+
+    source_object_tags = get_object_tags(
+        source_object_id,
+        include_deleted=include_deleted,
+        object_tag_class=object_tag_class,
+    )
+    copied_tags = ObjectTagClass.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:
+            existing_object_tag = dest_object_has_tag(object_tag)
+            if existing_object_tag:
+                # Now this tag is copied
+                existing_object_tag.is_copied = True
+                existing_object_tag.save()
+            else:
+                new_object_tag = ObjectTagClass()
+                new_object_tag.copy(object_tag)
+                new_object_tag.id = None  # To create a new instance in DB
+                new_object_tag.object_id = dest_object_id
+                new_object_tag.is_copied = True
+                new_object_tag.save()

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"),
+        ),
+    ]

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

openedx_tagging/core/tagging/rest_api/v1/serializers.py

@@ -176,6 +176,7 @@ def to_representation(self, instance: list[ObjectTag]) -> dict:
         """
         Convert this list of ObjectTags to the serialized dictionary, grouped by Taxonomy
         """
+        ObjectTagViewMinimalSerializer = self.context["view"].minimal_serilizer_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 +195,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
 
 

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,
@@ -444,6 +445,7 @@ class ObjectTagView(
     """
 
     serializer_class = ObjectTagSerializer
+    minimal_serilizer_class = ObjectTagMinimalSerializer
     permission_classes = [ObjectTagObjectPermissions]
     lookup_field = "object_id"
 

tests/openedx_tagging/core/tagging/test_api.py

@@ -913,3 +913,183 @@ 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_deleted(self) -> None:
+        obj1 = "object_id1"
+        obj2 = "object_id2"
+
+        tags_list = [
+            {
+                "value": "English",
+                "taxonomy": self.language_taxonomy,
+                "deleted": False,
+            },
+            {
+                "value": "DPANN",
+                "taxonomy": self.taxonomy,
+                "deleted": True,
+            },
+        ]
+
+        for tag_object in tags_list:
+            tagging_api.tag_object(object_id=obj1, taxonomy=tag_object["taxonomy"], tags=[tag_object["value"]])
+
+        tagging_api.delete_tags_from_taxonomy(self.taxonomy, ["DPANN"], with_subtags=True)
+
+        # Copy tags, also with deleted tags
+        tagging_api.copy_tags(obj1, obj2)
+
+        object_tags = tagging_api.get_object_tags(obj2, include_deleted=True)
+
+        assert len(object_tags) == 2
+        for index, object_tag in enumerate(object_tags):
+            assert object_tag.value == tags_list[index]["value"]
+            assert object_tag.is_deleted == tags_list[index]["deleted"]
+            assert object_tag.taxonomy == tags_list[index]["taxonomy"]
+            assert object_tag.object_id == obj2
+            assert object_tag.is_copied is True
+
+        # Copy tags, without deleted tags
+        tagging_api.copy_tags(obj1, obj2, include_deleted=False)
+
+        object_tags = tagging_api.get_object_tags(obj2, include_deleted=False)
+
+        assert len(object_tags) == 1
+        assert object_tags[0].value == tags_list[0]["value"]
+        assert not object_tags[0].is_deleted
+        assert object_tags[0].taxonomy == tags_list[0]["taxonomy"]
+        assert object_tags[0].object_id == obj2
+        assert object_tags[0].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"]