Pathway catalog and enrollment sink

README.rst

@@ -36,6 +36,12 @@ Below are the existing sink names and their corresponding object names (as neede
   model and stores the user profile data in ClickHouse. Object name: ``user_profile``
 - ``CourseEnrollmentSink`` - Listen for the ``ENROLL_STATUS_CHANGE`` event and stores
   the course enrollment data. Object name: ``course_enrollment``
+- ``CatalogPathwaySink`` - Listens for the ``post_save`` Django signal on the openedx-core
+  ``CatalogPathway`` model. Object name: ``catalog_pathway``
+- ``CatalogPathwayTombstoneSink`` - Listens for the ``post_delete`` Django signal on the
+  ``CatalogPathway`` model and marks the pathway deleted in ``catalog_pathway``.
+- ``PathwayEnrollmentSink`` - Listens for the ``post_save`` Django signal on the openedx-core
+  ``PathwayEnrollment`` model. Object name: ``pathway_enrollment``
 - ``UserRetirementSink`` - Listen for the ``USER_RETIRE_LMS_MISC`` Django signal and
   remove the user PII information from ClickHouse. This is a special sink with no assigned object name.
 

src/platform_plugin_aspects/settings/common.py

@@ -103,6 +103,14 @@ def plugin_settings(settings):
             "module": "openedx_tagging.models.base",
             "model": "ObjectTag",
         },
+        "catalog_pathway": {
+            "module": "openedx_catalog.models",
+            "model": "CatalogPathway",
+        },
+        "pathway_enrollment": {
+            "module": "openedx_catalog.models",
+            "model": "PathwayEnrollment",
+        },
     }
 
     # This is on by default here, off by default in tutor-contrib-aspects until we only

src/platform_plugin_aspects/signals.py

@@ -8,9 +8,12 @@
 from opaque_keys import InvalidKeyError
 
 from platform_plugin_aspects.sinks import (
+    CatalogPathwaySink,
+    CatalogPathwayTombstoneSink,
     CourseEnrollmentSink,
     ExternalIdSink,
     ObjectTagSink,
+    PathwayEnrollmentSink,
     TagSink,
     TaxonomySink,
     UserProfileSink,
@@ -303,3 +306,105 @@ def on_object_tag_deleted(  # pylint: disable=unused-argument  # pragma: no cove
 if _object_tag:  # pragma: no cover
     post_save.connect(on_object_tag_saved_txn, sender=_object_tag)
     post_delete.connect(on_object_tag_deleted, sender=_object_tag)
+
+
+def on_catalog_pathway_deleted(  # pylint: disable=unused-argument
+    sender, instance, **kwargs
+):
+    """
+    Receives post delete signal for CatalogPathway and queues a tombstone dump on commit.
+    """
+    # Django clears instance.pk after post_delete, so the id is read now, not in on_commit.
+    object_id = str(instance.id)
+
+    def queue_tombstone():
+        # 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 = CatalogPathwayTombstoneSink(None, None)
+        dump_data_to_clickhouse.delay(
+            sink_module=sink.__module__,
+            sink_name=sink.__class__.__name__,
+            object_id=object_id,
+        )
+
+    transaction.on_commit(queue_tombstone, robust=True)
+
+
+def on_catalog_pathway_saved_txn(*args, **kwargs):
+    """
+    Handle catalog_pathway saves in the middle of a transaction.
+
+    Handle saves in the middle of a transaction.
+    If this gets fired before the transaction commits, the task may try to
+    query an id that doesn't exist yet and throw an error. This should postpone
+    queuing the Celery task until after the transaction is committed.
+    """
+
+    def on_catalog_pathway_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 = CatalogPathwaySink(None, None)
+        dump_data_to_clickhouse.delay(
+            sink_module=sink.__module__,
+            sink_name=sink.__class__.__name__,
+            object_id=str(instance.id),
+        )
+
+    transaction.on_commit(lambda: on_catalog_pathway_saved(*args, **kwargs))
+
+
+# Connect the CatalogPathway.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_catalog_pathway = get_model("catalog_pathway")
+if _catalog_pathway:  # pragma: no cover
+    post_save.connect(on_catalog_pathway_saved_txn, sender=_catalog_pathway)
+    post_delete.connect(on_catalog_pathway_deleted, sender=_catalog_pathway)
+
+
+def on_pathway_enrollment_saved_txn(*args, **kwargs):
+    """
+    Handle pathway_enrollment saves in the middle of a transaction.
+
+    Handle saves in the middle of a transaction.
+    If this gets fired before the transaction commits, the task may try to
+    query an id that doesn't exist yet and throw an error. This should postpone
+    queuing the Celery task until after the transaction is committed.
+    """
+
+    def on_pathway_enrollment_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 = PathwayEnrollmentSink(None, None)
+        dump_data_to_clickhouse.delay(
+            sink_module=sink.__module__,
+            sink_name=sink.__class__.__name__,
+            object_id=str(instance.id),
+        )
+
+    transaction.on_commit(lambda: on_pathway_enrollment_saved(*args, **kwargs))
+
+
+# Connect the PathwayEnrollment.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_pathway_enrollment = get_model("pathway_enrollment")
+if _pathway_enrollment:  # pragma: no cover
+    post_save.connect(on_pathway_enrollment_saved_txn, sender=_pathway_enrollment)

src/platform_plugin_aspects/sinks/__init__.py

@@ -3,9 +3,11 @@
 """
 
 from .base_sink import BaseSink, ModelBaseSink
+from .catalog_pathway_sink import CatalogPathwaySink, CatalogPathwayTombstoneSink
 from .course_enrollment_sink import CourseEnrollmentSink
 from .course_overview_sink import CourseOverviewSink, XBlockSink
 from .external_id_sink import ExternalIdSink
+from .pathway_enrollment_sink import PathwayEnrollmentSink
 from .tag_sink import ObjectTagSink, TagSink, TaxonomySink
 from .user_profile_sink import UserProfileSink
 from .user_retire_sink import UserRetirementSink

src/platform_plugin_aspects/sinks/catalog_pathway_sink.py

@@ -0,0 +1,80 @@
+"""Catalog Pathway sink"""
+
+import json
+
+import requests
+
+from platform_plugin_aspects.sinks.base_sink import ModelBaseSink
+from platform_plugin_aspects.sinks.serializers import CatalogPathwaySerializer
+
+
+class CatalogPathwaySink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for the openedx-core CatalogPathway model
+    """
+
+    model = "catalog_pathway"
+    unique_key = "id"
+    clickhouse_table_name = "catalog_pathway"
+    timestamp_field = "time_last_dumped"
+    name = "Catalog Pathway"
+    serializer_class = CatalogPathwaySerializer
+
+    # Unlike the other sink tables, time_last_dumped is DateTime64 here because it is the
+    # ReplacingMergeTree version column; the serializer writes it with a UTC offset, which
+    # only "best_effort" parses.
+    CLICKHOUSE_BULK_INSERT_PARAMS = {
+        **ModelBaseSink.CLICKHOUSE_BULK_INSERT_PARAMS,
+        "date_time_input_format": "best_effort",
+    }
+
+    def get_queryset(self, start_pk=None):
+        return super().get_queryset(start_pk).select_related("org", "category")
+
+
+# pylint: disable-next=abstract-method
+class CatalogPathwayTombstoneSink(CatalogPathwaySink):
+    """
+    Sink for deleted openedx-core CatalogPathways
+
+    The pathway is gone from MySQL by the time this runs, so instead of serializing it, the
+    pathway's latest row in ClickHouse is copied with is_deleted set. A pathway that was never
+    dumped gets no row.
+    """
+
+    name = "Catalog Pathway Tombstone"
+
+    def dump(self, item_id, many=False, initial=None):
+        """
+        Write a tombstone row for the deleted pathway with id item_id.
+        """
+        table = f"{self.ch_database}.{self.clickhouse_table_name}"
+        params = {
+            "query": (
+                f"INSERT INTO {table} "
+                "SELECT * REPLACE (1 AS is_deleted, generateUUIDv4() AS dump_id, "
+                "now64(6) AS time_last_dumped) "
+                # Picked in a subquery: next to the REPLACE, ORDER BY time_last_dumped would
+                # resolve to the now64(6) alias and copy an arbitrary version.
+                f"FROM (SELECT * FROM {table} WHERE id = {int(item_id)} "
+                "ORDER BY time_last_dumped DESC LIMIT 1)"
+            ),
+        }
+        request = requests.Request(
+            "POST",
+            self.ch_url,
+            params=params,
+            auth=self.ch_auth,
+        )
+        self.log.info(f"Now dumping {self.name} {item_id} to ClickHouse")
+        # Retried once: the pathway is gone from MySQL, so a lost tombstone can't be backfilled.
+        try:
+            response = self._send_clickhouse_request(request)
+        except requests.RequestException:
+            response = self._send_clickhouse_request(request)
+        summary = json.loads(response.headers.get("X-ClickHouse-Summary", "{}"))
+        if summary.get("written_rows") == "0":
+            self.log.warning(
+                f"No {self.name} written for {item_id}, it was never dumped"
+            )
+        self.log.info(f"Completed dumping {self.name} {item_id} to ClickHouse")

src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py

@@ -0,0 +1,20 @@
+"""Pathway Enrollment sink"""
+
+from platform_plugin_aspects.sinks.base_sink import ModelBaseSink
+from platform_plugin_aspects.sinks.serializers import PathwayEnrollmentSerializer
+
+
+class PathwayEnrollmentSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for the openedx-core PathwayEnrollment model
+    """
+
+    model = "pathway_enrollment"
+    unique_key = "id"
+    clickhouse_table_name = "pathway_enrollment"
+    timestamp_field = "time_last_dumped"
+    name = "Pathway Enrollment"
+    serializer_class = PathwayEnrollmentSerializer
+
+    def get_queryset(self, start_pk=None):
+        return super().get_queryset(start_pk).select_related("user")

src/platform_plugin_aspects/sinks/serializers.py

@@ -8,7 +8,11 @@
 from pytz import UTC
 from rest_framework import serializers
 
-from platform_plugin_aspects.utils import get_model, get_tags_for_block
+from platform_plugin_aspects.utils import (
+    get_model,
+    get_or_create_xapi_actor_id,
+    get_tags_for_block,
+)
 
 
 class DateTimeJSONEncoder(json.JSONEncoder):
@@ -207,9 +211,74 @@
 
     def get_course_key(self, obj):
         """Return the course key as a string."""
         return str(obj.course_id)
 
 
+class CatalogPathwaySerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """Serializer for the CatalogPathway model."""
+
+    org = serializers.CharField(source="org.short_name")
+    category_code = serializers.CharField(source="category.category_code")
+    category_title = serializers.CharField(source="category.title")
+    is_deleted = serializers.SerializerMethodField()
+
+    class Meta:
+        """Meta class for the CatalogPathwaySerializer"""
+
+        model = get_model("catalog_pathway")
+        fields = [
+            "id",
+            "org",
+            "pathway_code",
+            "title",
+            "category_code",
+            "category_title",
+            "created",
+            "modified",
+            "dump_id",
+            "time_last_dumped",
+            "is_deleted",
+        ]
+
+    def get_is_deleted(self, instance):  # pylint: disable=unused-argument
+        """
+        Saved pathways are never deleted; CatalogPathwayTombstoneSink writes deletions.
+
+        0 rather than False: the column is the UInt8 is_deleted of a ReplacingMergeTree, and
+        ClickHouse won't parse a CSV "False" as a UInt8.
+        """
+        return 0
+
+
+class PathwayEnrollmentSerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """
+    Serializer for the PathwayEnrollment model.
+
+    The learner is stored as their xAPI actor id instead of user_id to keep the table PII free.
+    """
+
+    actor_id = serializers.SerializerMethodField()
+
+    class Meta:
+        """Meta class for the PathwayEnrollmentSerializer"""
+
+        model = get_model("pathway_enrollment")
+        fields = [
+            "id",
+            "catalog_pathway_id",
+            "actor_id",
+            "created",
+            "modified",
+            "is_active",
+            "dump_id",
+            "time_last_dumped",
+        ]
+
+    def get_actor_id(self, instance):
+        """Return the learner's xAPI actor id instead of any user identifier."""
+        return get_or_create_xapi_actor_id(instance.user)
+
+
 class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer):
     """Serializer for the Tag model."""