feat: add student competency status sinks

README.rst

@@ -42,6 +42,13 @@ Below are the existing sink names and their corresponding object names (as neede
   ``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``
+- ``StudentCompetencyStatusSink`` - Listens for the ``post_save`` Django signal on the openedx-core
+  ``StudentCompetencyStatus`` model (a learner's mastery status for one competency). Object name:
+  ``student_competency_status``
+- ``StudentCompetencyCriteriaGroupStatusSink`` - Listens for the ``post_save`` Django signal on the
+  openedx-core ``StudentCompetencyCriteriaGroupStatus`` model (a learner's mastery status for one
+  criteria group, with the group's competency, course and level). Object name:
+  ``student_competency_criteria_group_status``
 - ``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

@@ -111,6 +111,14 @@ def plugin_settings(settings):
             "module": "openedx_catalog.models",
             "model": "PathwayEnrollment",
         },
+        "student_competency_status": {
+            "module": "openedx_learning.models",
+            "model": "StudentCompetencyStatus",
+        },
+        "student_competency_criteria_group_status": {
+            "module": "openedx_learning.models",
+            "model": "StudentCompetencyCriteriaGroupStatus",
+        },
     }
 
     # This is on by default here, off by default in tutor-contrib-aspects until we only

src/platform_plugin_aspects/signals.py

@@ -14,6 +14,8 @@
     ExternalIdSink,
     ObjectTagSink,
     PathwayEnrollmentSink,
+    StudentCompetencyCriteriaGroupStatusSink,
+    StudentCompetencyStatusSink,
     TagSink,
     TaxonomySink,
     UserProfileSink,
@@ -408,3 +410,89 @@ def on_pathway_enrollment_saved(  # pylint: disable=unused-argument  # pragma: n
 _pathway_enrollment = get_model("pathway_enrollment")
 if _pathway_enrollment:  # pragma: no cover
     post_save.connect(on_pathway_enrollment_saved_txn, sender=_pathway_enrollment)
+
+
+def on_student_competency_status_saved_txn(*args, **kwargs):
+    """
+    Handle student_competency_status saves in the middle of a transaction.
+
+    The status is written inside the LMS grade transaction, so queuing the dump
+    before the commit could read a row that doesn't exist yet. This postpones
+    queuing the Celery task until after the transaction is committed.
+
+    openedx-core raises a stored status with a bulk UPDATE and sends post_save
+    itself, so raises arrive here as well as creates.
+    """
+
+    def on_student_competency_status_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 = StudentCompetencyStatusSink(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_student_competency_status_saved(*args, **kwargs))
+
+
+# Connect the StudentCompetencyStatus.post_save signal handler only if we have a model to attach to.
+# (prevents celery errors during tests)
+_student_competency_status = get_model("student_competency_status")
+if _student_competency_status:  # pragma: no cover
+    post_save.connect(
+        on_student_competency_status_saved_txn, sender=_student_competency_status
+    )
+
+
+def on_student_competency_criteria_group_status_saved_txn(*args, **kwargs):
+    """
+    Handle student_competency_criteria_group_status saves in the middle of a transaction.
+
+    Group statuses are written by the rollup, inside a transaction per level, so
+    queuing the dump before the commit could read a row that doesn't exist yet.
+    openedx-core sends post_save itself when it raises a stored status.
+    """
+
+    def on_student_competency_criteria_group_status_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 = StudentCompetencyCriteriaGroupStatusSink(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_student_competency_criteria_group_status_saved(*args, **kwargs)
+    )
+
+
+# Connect the StudentCompetencyCriteriaGroupStatus.post_save signal handler only if we have a
+# model to attach to. (prevents celery errors during tests)
+_student_competency_criteria_group_status = get_model(
+    "student_competency_criteria_group_status"
+)
+if _student_competency_criteria_group_status:  # pragma: no cover
+    post_save.connect(
+        on_student_competency_criteria_group_status_saved_txn,
+        sender=_student_competency_criteria_group_status,
+    )

src/platform_plugin_aspects/sinks/__init__.py

@@ -8,6 +8,10 @@
 from .course_overview_sink import CourseOverviewSink, XBlockSink
 from .external_id_sink import ExternalIdSink
 from .pathway_enrollment_sink import PathwayEnrollmentSink
+from .student_competency_status_sink import (
+    StudentCompetencyCriteriaGroupStatusSink,
+    StudentCompetencyStatusSink,
+)
 from .tag_sink import ObjectTagSink, TagSink, TaxonomySink
 from .user_profile_sink import UserProfileSink
 from .user_retire_sink import UserRetirementSink

src/platform_plugin_aspects/sinks/serializers.py

@@ -279,6 +279,105 @@ def get_actor_id(self, instance):
         return get_or_create_xapi_actor_id(instance.user)
 
 
+class StudentCompetencyStatusSerializer(
+    BaseSinkSerializer, serializers.ModelSerializer
+):
+    """
+    Serializer for the openedx-core StudentCompetencyStatus model.
+
+    Statuses only rise, so ``created`` is when the learner first got a status for the
+    competency and ``modified`` on a Demonstrated row is when they demonstrated it.
+    """
+
+    actor_id = serializers.SerializerMethodField()
+    taxonomy_id = serializers.IntegerField(source="tag.taxonomy_id")
+    status = serializers.CharField(source="status.status")
+
+    class Meta:
+        """Meta class for the StudentCompetencyStatusSerializer"""
+
+        model = get_model("student_competency_status")
+        fields = [
+            "id",
+            "actor_id",
+            "tag_id",
+            "taxonomy_id",
+            "status_id",
+            "status",
+            "created",
+            "modified",
+            "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 StudentCompetencyCriteriaGroupStatusSerializer(
+    BaseSinkSerializer, serializers.ModelSerializer
+):
+    """
+    Serializer for the openedx-core StudentCompetencyCriteriaGroupStatus model.
+
+    The group's competency, course and level are denormalized so ClickHouse needs no
+    criteria-group table: ``group_level`` is ``root`` (no parent), ``course`` (the
+    group scoping a competency to one course run) or ``leaf`` (below a course group).
+    ``course_key`` is the group's own course run, or for a leaf, its course group's.
+    """
+
+    actor_id = serializers.SerializerMethodField()
+    group_id = serializers.IntegerField(source="group.id")
+    parent_group_id = serializers.IntegerField(source="group.parent_id", allow_null=True)
+    tag_id = serializers.IntegerField(source="group.tag_id")
+    taxonomy_id = serializers.IntegerField(source="group.tag.taxonomy_id")
+    course_key = serializers.SerializerMethodField()
+    group_level = serializers.SerializerMethodField()
+    status = serializers.CharField(source="status.status")
+
+    class Meta:
+        """Meta class for the StudentCompetencyCriteriaGroupStatusSerializer"""
+
+        model = get_model("student_competency_criteria_group_status")
+        fields = [
+            "id",
+            "actor_id",
+            "group_id",
+            "parent_group_id",
+            "tag_id",
+            "taxonomy_id",
+            "course_key",
+            "group_level",
+            "status_id",
+            "status",
+            "created",
+            "modified",
+            "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)
+
+    def get_group_level(self, instance):
+        """Return root, course or leaf for the group."""
+        group = instance.group
+        if group.parent_id is None:
+            return "root"
+        if group.course_id is not None:
+            return "course"
+        return "leaf"
+
+    def get_course_key(self, instance):
+        """Return the course key of the group's course run, or its nearest ancestor's; empty for a root."""
+        group = instance.group
+        while group is not None and group.course_id is None:
+            group = group.parent
+        return str(group.course.course_key) if group is not None else ""
+
+
 class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer):
     """Serializer for the Tag model."""
 

src/platform_plugin_aspects/sinks/student_competency_status_sink.py

@@ -0,0 +1,45 @@
+"""Student Competency Status sinks: per competency and per criteria group"""
+
+from platform_plugin_aspects.sinks.base_sink import ModelBaseSink
+from platform_plugin_aspects.sinks.serializers import (
+    StudentCompetencyCriteriaGroupStatusSerializer,
+    StudentCompetencyStatusSerializer,
+)
+
+
+class StudentCompetencyStatusSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for the openedx-core StudentCompetencyStatus model: a learner's current mastery status for one competency
+    """
+
+    model = "student_competency_status"
+    unique_key = "id"
+    clickhouse_table_name = "student_competency_status"
+    timestamp_field = "time_last_dumped"
+    name = "Student Competency Status"
+    serializer_class = StudentCompetencyStatusSerializer
+
+    def get_queryset(self, start_pk=None):
+        return super().get_queryset(start_pk).select_related("user", "tag", "status")
+
+
+class StudentCompetencyCriteriaGroupStatusSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for the openedx-core StudentCompetencyCriteriaGroupStatus model: a learner's current
+    mastery status for one criteria group, which is the only place AttemptedNotDemonstrated is
+    stored and, for course-level groups, the learner's status in one course
+    """
+
+    model = "student_competency_criteria_group_status"
+    unique_key = "id"
+    clickhouse_table_name = "student_competency_criteria_group_status"
+    timestamp_field = "time_last_dumped"
+    name = "Student Competency Criteria Group Status"
+    serializer_class = StudentCompetencyCriteriaGroupStatusSerializer
+
+    def get_queryset(self, start_pk=None):
+        return (
+            super()
+            .get_queryset(start_pk)
+            .select_related("user", "status", "group__tag", "group__course")
+        )

src/platform_plugin_aspects/sinks/tests/test_serializers.py

@@ -10,6 +10,8 @@
     CourseOverviewSerializer,
     DateTimeJSONEncoder,
     PathwayEnrollmentSerializer,
+    StudentCompetencyCriteriaGroupStatusSerializer,
+    StudentCompetencyStatusSerializer,
 )
 from test_utils.helpers import course_key_factory
 
@@ -132,3 +134,78 @@ def test_get_actor_id(self, mock_get_or_create_xapi_actor_id):
 
         assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18"
         mock_get_or_create_xapi_actor_id.assert_called_once_with(enrollment.user)
+
+
+class TestStudentCompetencyStatusSerializer(TestCase):
+    """
+    Test StudentCompetencyStatusSerializer
+    """
+
+    @patch("platform_plugin_aspects.sinks.serializers.get_or_create_xapi_actor_id")
+    def test_get_actor_id(self, mock_get_xapi_actor_id):
+        """
+        Test that the learner is serialized as their xAPI actor id.
+        """
+        mock_get_xapi_actor_id.return_value = "29e5b22b-b457-49fa-9633-6c0123305d18"
+        status = Mock()
+
+        actor_id = StudentCompetencyStatusSerializer().get_actor_id(status)
+
+        assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18"
+        mock_get_xapi_actor_id.assert_called_once_with(status.user)
+
+
+class TestStudentCompetencyCriteriaGroupStatusSerializer(TestCase):
+    """
+    Test StudentCompetencyCriteriaGroupStatusSerializer
+    """
+
+    def setUp(self):
+        self.serializer = StudentCompetencyCriteriaGroupStatusSerializer()
+        # "parent" is a Mock constructor argument, so it is set as an attribute.
+        self.root = Mock(parent_id=None, course_id=None)
+        self.root.parent = None
+        self.course_group = Mock(parent_id=1, course_id=7)
+        self.course_group.parent = self.root
+        self.course_group.course.course_key = "course-v1:Org+Course+Run"
+        self.leaf = Mock(parent_id=2, course_id=None)
+        self.leaf.parent = self.course_group
+
+    def test_group_level(self):
+        """
+        Test that groups are serialized as root, course or leaf.
+        """
+        levels = [
+            self.serializer.get_group_level(Mock(group=group))
+            for group in (self.root, self.course_group, self.leaf)
+        ]
+
+        assert levels == ["root", "course", "leaf"]
+
+    def test_course_key(self):
+        """
+        Test that a leaf takes its course group's course key and a root has none.
+        """
+        assert (
+            self.serializer.get_course_key(Mock(group=self.course_group))
+            == "course-v1:Org+Course+Run"
+        )
+        assert (
+            self.serializer.get_course_key(Mock(group=self.leaf))
+            == "course-v1:Org+Course+Run"
+        )
+        assert self.serializer.get_course_key(Mock(group=self.root)) == ""
+
+    @patch("platform_plugin_aspects.sinks.serializers.get_or_create_xapi_actor_id")
+    def test_get_actor_id(self, mock_get_or_create_xapi_actor_id):
+        """
+        Test that the learner is serialized as their xAPI actor id.
+        """
+        mock_get_or_create_xapi_actor_id.return_value = "29e5b22b-b457-49fa-9633-6c0123305d18"
+        status = Mock()
+
+        assert (
+            self.serializer.get_actor_id(status)
+            == "29e5b22b-b457-49fa-9633-6c0123305d18"
+        )
+        mock_get_or_create_xapi_actor_id.assert_called_once_with(status.user)