feat: add course enrollment sink

README.rst

@@ -23,6 +23,8 @@ Available Sinks
   model and stores the user profile data in ClickHouse.
 - `UserRetirementSink` - Listen for the `USER_RETIRE_LMS_MISC` Django signal and
   remove the user PII information from ClickHouse.
+- `CourseEnrollmentSink` - Listen for the `ENROLL_STATUS_CHANGE` event and stores
+  the course enrollment data.
 
 Commands
 ========

platform_plugin_aspects/apps.py

@@ -49,7 +49,18 @@ class PlatformPluginAspectsConfig(AppConfig):
                         PluginSignals.SIGNAL_PATH: "xmodule.modulestore.django.COURSE_PUBLISHED",
                     }
                 ],
-            }
+            },
+            "lms.djangoapp": {
+                # List of all plugin Signal receivers for this app and project type.
+                PluginSignals.RECEIVERS: [
+                    {
+                        # The name of the app's signal receiver function.
+                        PluginSignals.RECEIVER_FUNC_NAME: "receive_course_enrollment_changed",
+                        # The full path to the module where the signal is defined.
+                        PluginSignals.SIGNAL_PATH: "common.djangoapps.student.signals.signals.ENROLL_STATUS_CHANGE",
+                    }
+                ],
+            },
         },
     }
 

platform_plugin_aspects/settings/common.py

@@ -82,6 +82,10 @@ def plugin_settings(settings):
             "module": "openedx.core.djangoapps.external_user_ids.models",
             "model": "ExternalId",
         },
+        "course_enrollment": {
+            "module": "common.djangoapps.student.models",
+            "model": "CourseEnrollment",
+        },
         "custom_course_edx": {
             "module": "lms.djangoapps.ccx.models",
             "model": "CustomCourseForEdX",

platform_plugin_aspects/signals.py

@@ -7,6 +7,7 @@
 from django.dispatch import Signal, receiver
 
 from platform_plugin_aspects.sinks import (
+    CourseEnrollmentSink,
     ExternalIdSink,
     UserProfileSink,
     UserRetirementSink,
@@ -34,6 +35,31 @@ def receive_course_publish(  # pylint: disable=unused-argument  # pragma: no cov
     dump_course_to_clickhouse.delay(str(course_key))
 
 
+def receive_course_enrollment_changed(  # pylint: disable=unused-argument  # pragma: no cover
+    sender, **kwargs
+):
+    """
+    Receives ENROLL_STATUS_CHANGE
+    """
+    from platform_plugin_aspects.tasks import (  # pylint: disable=import-outside-toplevel
+        dump_data_to_clickhouse,
+    )
+
+    user = kwargs.get("user")
+    course_id = kwargs.get("course_id")
+
+    CourseEnrollment = get_model("course_enrollment")
+    instance = CourseEnrollment.objects.get(user=user, course_id=course_id)
+
+    sink = CourseEnrollmentSink(None, None)
+
+    dump_data_to_clickhouse.delay(
+        sink_module=sink.__module__,
+        sink_name=sink.__class__.__name__,
+        object_id=instance.id,
+    )
+
+
 def on_user_profile_updated(instance):
     """
     Queues the UserProfile dump job when the parent transaction is committed.

platform_plugin_aspects/sinks/__init__.py

@@ -3,6 +3,7 @@
 """
 
 from .base_sink import BaseSink, ModelBaseSink
+from .course_enrollment_sink import CourseEnrollmentSink
 from .course_overview_sink import CourseOverviewSink, XBlockSink
 from .external_id_sink import ExternalIdSink
 from .user_profile_sink import UserProfileSink

platform_plugin_aspects/sinks/base_sink.py

@@ -271,7 +271,7 @@ def send_item(self, serialized_item, many=False):
                 writer.writerow(node.values())
         else:
             writer.writerow(serialized_item.values())
-
+        print("ClickHouse CSV output", output.getvalue().encode("utf-8"))
         request = requests.Request(
             "POST",
             self.ch_url,

platform_plugin_aspects/sinks/course_enrollment_sink.py

@@ -0,0 +1,20 @@
+"""User profile sink"""
+
+from platform_plugin_aspects.sinks.base_sink import ModelBaseSink
+from platform_plugin_aspects.sinks.serializers import CourseEnrollmentSerializer
+
+
+class CourseEnrollmentSink(ModelBaseSink):  # pylint: disable=abstract-method
+    """
+    Sink for user CourseEnrollment model
+    """
+
+    model = "course_enrollment"
+    unique_key = "id"
+    clickhouse_table_name = "course_enrollment"
+    timestamp_field = "time_last_dumped"
+    name = "Course Enrollment"
+    serializer_class = CourseEnrollmentSerializer
+
+    def get_queryset(self, start_pk=None):
+        return super().get_queryset(start_pk).select_related("user")

platform_plugin_aspects/sinks/serializers.py

@@ -179,3 +179,30 @@ def get_course_data_json(self, overview):
     def get_course_key(self, overview):
         """Return the course key as a string."""
         return str(overview.id)
+
+
+class CourseEnrollmentSerializer(BaseSinkSerializer, serializers.ModelSerializer):
+    """Serializer for the Course Enrollment model."""
+
+    course_key = serializers.SerializerMethodField()
+    username = serializers.CharField(source="user.username")
+
+    class Meta:
+        """Meta class for the CourseEnrollmentSerializer"""
+
+        model = get_model("course_enrollment")
+        fields = [
+            "id",
+            "course_key",
+            "created",
+            "is_active",
+            "mode",
+            "username",
+            "user_id",
+            "dump_id",
+            "time_last_dumped",
+        ]
+
+    def get_course_key(self, obj):
+        """Return the course key as a string."""
+        return str(obj.course_id)