feat: add tag and taxonomy dump

fix: add lineage information for tag sink

chore: quality fixes

fix: serialize lineage data as json

Author: Ian2012

platform_plugin_aspects/settings/common.py

@@ -94,4 +94,13 @@ def plugin_settings(settings):
             "module": "openedx.core.djangoapps.user_api.models",
             "model": "UserPreference",
         },
+        "tag": {"module": "openedx_tagging.core.tagging.models", "model": "Tag"},
+        "taxonomy": {
+            "module": "openedx_tagging.core.tagging.models",
+            "model": "Taxonomy",
+        },
+        "object_tag": {
+            "module": "openedx_tagging.core.tagging.models",
+            "model": "ObjectTag",
+        },
     }

platform_plugin_aspects/signals.py

@@ -9,6 +9,9 @@
 from platform_plugin_aspects.sinks import (
     CourseEnrollmentSink,
     ExternalIdSink,
+    ObjectTagSink,
+    TagSink,
+    TaxonomySink,
     UserProfileSink,
     UserRetirementSink,
 )
@@ -143,3 +146,81 @@ def on_user_retirement(  # pylint: disable=unused-argument  # pragma: no cover
         sink_name=sink.__class__.__name__,
         object_id=str(user.id),
     )
+
+
+def on_tag_saved(  # pylint: disable=unused-argument  # pragma: no cover
+    sender, instance, **kwargs
+):
+    """
+    Receives post save signal and queues the dump job.
+    """
+    # import here, because signal is registered at startup, but items in tasks are not yet able to be loaded
+    from platform_plugin_aspects.tasks import (  # pylint: disable=import-outside-toplevel
+        dump_data_to_clickhouse,
+    )
+
+    sink = TagSink(None, None)
+    dump_data_to_clickhouse.delay(
+        sink_module=sink.__module__,
+        sink_name=sink.__class__.__name__,
+        object_id=str(instance.id),
+    )
+
+
+# Connect the ExternalId.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_tag = get_model("tag")
+if _tag:
+    post_save.connect(on_tag_saved, sender=_tag)  # pragma: no cover
+
+
+def on_taxonomy_saved(  # pylint: disable=unused-argument  # pragma: no cover
+    sender, instance, **kwargs
+):
+    """
+    Receives post save signal and queues the dump job.
+    """
+    # import here, because signal is registered at startup, but items in tasks are not yet able to be loaded
+    from platform_plugin_aspects.tasks import (  # pylint: disable=import-outside-toplevel
+        dump_data_to_clickhouse,
+    )
+
+    sink = TaxonomySink(None, None)
+    dump_data_to_clickhouse.delay(
+        sink_module=sink.__module__,
+        sink_name=sink.__class__.__name__,
+        object_id=str(instance.id),
+    )
+
+
+# Connect the ExternalId.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_taxonomy = get_model("taxonomy")
+if _taxonomy:
+    post_save.connect(on_taxonomy_saved, sender=_taxonomy)  # pragma: no cover
+
+
+def on_object_tag_saved(  # pylint: disable=unused-argument  # pragma: no cover
+    sender, instance, **kwargs
+):
+    """
+    Receives post save signal and queues the dump job.
+    """
+    # import here, because signal is registered at startup, but items in tasks are not yet able to be loaded
+    from platform_plugin_aspects.tasks import (  # pylint: disable=import-outside-toplevel
+        dump_data_to_clickhouse,
+    )
+
+    sink = ObjectTagSink(None, None)
+    dump_data_to_clickhouse.delay(
+        sink_module=sink.__module__,
+        sink_name=sink.__class__.__name__,
+        object_id=str(instance.id),
+    )
+
+
+# Connect the ExternalId.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_object_tag = get_model("object_tag")
+if _object_tag:
+    post_save.connect(on_object_tag_saved, sender=_object_tag)  # pragma: no cover

platform_plugin_aspects/sinks/__init__.py

@@ -6,5 +6,6 @@
 from .course_enrollment_sink import CourseEnrollmentSink
 from .course_overview_sink import CourseOverviewSink, XBlockSink
 from .external_id_sink import ExternalIdSink
+from .tag_sink import ObjectTagSink, TagSink, TaxonomySink
 from .user_profile_sink import UserProfileSink
 from .user_retire_sink import UserRetirementSink

platform_plugin_aspects/sinks/serializers.py

@@ -206,3 +206,67 @@ class Meta:
     def get_course_key(self, obj):
         """Return the course key as a string."""
         return str(obj.course_id)
+
+
+class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """Serializer for the Tag model."""
+
+    lineage = serializers.SerializerMethodField()
+
+    class Meta:
+        """Meta class for the TagSerializer."""
+
+        model = get_model("tag")
+        fields = [
+            "id",
+            "taxonomy",
+            "parent",
+            "value",
+            "external_id",
+            "lineage",
+            "dump_id",
+            "time_last_dumped",
+        ]
+
+    def get_lineage(self, instance):
+        return json.dumps(instance.get_lineage())
+
+
+class ObjectTagSerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """Serializer for the ObjectTag model."""
+
+    lineage = serializers.SerializerMethodField()
+
+    class Meta:
+        """Meta class for the ObjectTagSerializer"""
+
+        model = get_model("object_tag")
+        fields = [
+            "id",
+            "object_id",
+            "taxonomy",
+            "tag",
+            "_value",
+            "_export_id",
+            "lineage",
+            "dump_id",
+            "time_last_dumped",
+        ]
+
+    def get_lineage(self, instance):
+        return json.dumps(instance.get_lineage())
+
+
+class TaxonomySerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """Serializer for the Taxonomy model."""
+
+    class Meta:
+        """Meta class for the TaxonomySerializer."""
+
+        model = get_model("taxonomy")
+        fields = [
+            "id",
+            "name",
+            "dump_id",
+            "time_last_dumped",
+        ]

platform_plugin_aspects/sinks/tag_sink.py

@@ -0,0 +1,47 @@
+"""Tag sink"""
+
+from platform_plugin_aspects.sinks.base_sink import ModelBaseSink
+from platform_plugin_aspects.sinks.serializers import (
+    ObjectTagSerializer,
+    TagSerializer,
+    TaxonomySerializer,
+)
+
+
+class TagSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for content tags
+    """
+
+    model = "tag"
+    unique_key = "id"
+    clickhouse_table_name = "tag"
+    timestamp_field = "time_last_dumped"
+    name = "Tag"
+    serializer_class = TagSerializer
+
+
+class TaxonomySink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for content taxonomy
+    """
+
+    model = "taxonomy"
+    unique_key = "id"
+    clickhouse_table_name = "taxonomy"
+    timestamp_field = "time_last_dumped"
+    name = "Taxonomy"
+    serializer_class = TaxonomySerializer
+
+
+class ObjectTagSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for tagged objects
+    """
+
+    model = "object_tag"
+    unique_key = "id"
+    clickhouse_table_name = "object_tag"
+    timestamp_field = "time_last_dumped"
+    name = "ObjectTag"
+    serializer_class = ObjectTagSerializer

platform_plugin_aspects/tasks.py

@@ -59,3 +59,6 @@ def dump_data_to_clickhouse(
     if Sink.is_enabled():
         sink = Sink(connection_overrides=connection_overrides, log=celery_log)
         sink.dump(object_id)
+        return "Dumped"
+
+    return "Disabled"