feat: adds xApiTransforms for completion aggregator events

CHANGELOG.rst

@@ -14,6 +14,12 @@ Change Log
 Unreleased
 ~~~~~~~~~~
 
+[4.0.4] - 2024-06-13
+~~~~~~~~~~~~~~~~~~~~
+
+* Emit tracking log events for `openedx.completion_aggregator.progress.*` and
+  `openedx.completion_aggregator.completion.*` for the various block/course types
+
 [4.0.3] - 2023-10-24
 ~~~~~~~~~~~~~~~~~~~~
 

completion_aggregator/__init__.py

@@ -5,4 +5,4 @@
 
 from __future__ import absolute_import, unicode_literals
 
-__version__ = '4.0.3'
+__version__ = '4.0.4'

completion_aggregator/apps.py

@@ -38,6 +38,10 @@ def ready(self):
         """
         Load signal handlers when the app is ready.
         """
+        # pylint: disable=import-outside-toplevel
         from . import signals
         signals.register()
-        from .tasks import aggregation_tasks, handler_tasks  # pylint: disable=unused-import
+
+        # pylint: disable=unused-import
+        from . import transformers
+        from .tasks import aggregation_tasks, handler_tasks

completion_aggregator/models.py

@@ -5,6 +5,7 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from eventtracking import tracker
 from opaque_keys.edx.django.models import CourseKeyField, UsageKeyField
 from opaque_keys.edx.keys import CourseKey, UsageKey
 
@@ -171,8 +172,44 @@ def submit_completion(self, user, course_key, block_key, aggregation_name, earne
                 'last_modified': last_modified,
             },
         )
+        self.emit_completion_aggregator_logs([obj])
+
         return obj, is_new
 
+    @staticmethod
+    def emit_completion_aggregator_logs(updated_aggregators):
+        """
+        Emit a tracking log for each element of the list parameter.
+
+        Parameters
+        ----------
+            updated_aggregators: List of Aggregator intances
+
+        """
+        for aggregator in updated_aggregators:
+            event = "progress" if aggregator.percent < 1 else "completion"
+            event_type = aggregator.aggregation_name
+
+            if event_type not in settings.ALLOWED_COMPLETION_AGGREGATOR_EVENT_TYPES.get(event, {}):
+                continue
+
+            event_name = f"openedx.completion_aggregator.{event}.{event_type}"
+
+            tracker.emit(
+                event_name,
+                {
+                    "user_id": aggregator.user_id,
+                    "course_id": str(aggregator.course_key),
+                    "block_id": str(aggregator.block_key),
+                    "modified": aggregator.modified,
+                    "created": aggregator.created,
+                    "earned": aggregator.earned,
+                    "possible": aggregator.possible,
+                    "percent": aggregator.percent,
+                    "type": event_type,
+                }
+            )
+
     def bulk_create_or_update(self, updated_aggregators):
         """
         Update the collection of aggregator object using mysql insert on duplicate update query.
@@ -194,6 +231,7 @@ def bulk_create_or_update(self, updated_aggregators):
                 else:
                     aggregation_data = [obj.get_values() for obj in updated_aggregators]
                     cur.executemany(INSERT_OR_UPDATE_AGGREGATOR_QUERY, aggregation_data)
+                    self.emit_completion_aggregator_logs(updated_aggregators)
 
 
 class Aggregator(TimeStampedModel):

completion_aggregator/settings/aws.py

@@ -4,11 +4,16 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from event_routing_backends.settings import production as erb_settings
+
 
 def plugin_settings(settings):
     """
     Modify the provided settings object with settings specific to this plugin.
     """
+    # Load Event Routing Backend production settings first.
+    erb_settings.plugin_settings(settings)
+
     settings.COMPLETION_AGGREGATOR_BLOCK_TYPES = set(settings.ENV_TOKENS.get(
         'COMPLETION_AGGREGATOR_BLOCK_TYPES',
         settings.COMPLETION_AGGREGATOR_BLOCK_TYPES,

completion_aggregator/settings/common.py

@@ -4,17 +4,37 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from event_routing_backends.utils.settings import event_tracking_backends_config
+
 
 def plugin_settings(settings):
     """
     Modify the provided settings object with settings specific to this plugin.
     """
+    # Emit feature allows to publish two kind of events progress and completion
+    # This setting controls which type of event will be published to change the default behavior
+    # the block type should be removed or added from the progress or completion list.
+    settings.ALLOWED_COMPLETION_AGGREGATOR_EVENT_TYPES = {
+        "progress": {
+            "course",
+            "chapter",
+            "sequential",
+            "vertical",
+        },
+        "completion": {
+            "course",
+            "chapter",
+            "sequential",
+            "vertical",
+        }
+    }
     settings.COMPLETION_AGGREGATOR_BLOCK_TYPES = {
         'course',
         'chapter',
         'sequential',
         'vertical',
     }
+
     settings.COMPLETION_AGGREGATOR_ASYNC_AGGREGATION = False
 
     # Names of the batch operations locks
@@ -35,3 +55,16 @@ def plugin_settings(settings):
     # 1. All courses should be reaggregated for the changes to take effect.
     # 2. It's not possible to revert this change by reaggregation without manually removing existing Aggregators.
     settings.COMPLETION_AGGREGATOR_AGGREGATE_UNRELEASED_BLOCKS = False
+
+    # Whitelist the aggregator events for use with event routing backends xAPI backend.
+    enabled_aggregator_events = [
+        f'openedx.completion_aggregator.{event_type}.{block_type}'
+
+        for event_type in settings.ALLOWED_COMPLETION_AGGREGATOR_EVENT_TYPES
+        for block_type in settings.ALLOWED_COMPLETION_AGGREGATOR_EVENT_TYPES[event_type]
+    ]
+    settings.EVENT_TRACKING_BACKENDS_ALLOWED_XAPI_EVENTS += enabled_aggregator_events
+    settings.EVENT_TRACKING_BACKENDS.update(event_tracking_backends_config(
+        settings.EVENT_TRACKING_BACKENDS_ALLOWED_XAPI_EVENTS,
+        settings.EVENT_TRACKING_BACKENDS_ALLOWED_CALIPER_EVENTS,
+    ))

completion_aggregator/transformers.py

@@ -7,6 +7,11 @@
 except ImportError:
     BlockStructureTransformer = object
 
+from event_routing_backends.processors.openedx_filters.decorators import openedx_filter
+from event_routing_backends.processors.xapi import constants
+from event_routing_backends.processors.xapi.registry import XApiTransformersRegistry
+from event_routing_backends.processors.xapi.transformer import XApiTransformer
+from tincan import Activity, ActivityDefinition, LanguageMap, Result, Verb
 from xblock.completable import XBlockCompletionMode
 
 
@@ -79,3 +84,119 @@ def transform(self, usage_info, block_structure):  # pylint: disable=unused-argu
             if completion_mode != XBlockCompletionMode.EXCLUDED:
                 aggregators = self.calculate_aggregators(block_structure, block_key)
                 block_structure.set_transformer_block_field(block_key, self, self.AGGREGATORS, aggregators)
+
+
+class BaseAggregatorXApiTransformer(XApiTransformer):
+    """
+    Base transformer for all completion aggregator events.
+    """
+
+    object_type = None
+
+    def get_object(self) -> Activity:
+        """
+        Get object for xAPI transformed event.
+        """
+        if not self.object_type:
+            raise NotImplementedError()
+
+        return Activity(
+            id=self.get_object_iri("xblock", self.get_data("data.block_id")),
+            definition=ActivityDefinition(
+                type=self.object_type,
+            ),
+        )
+
+
+class BaseProgressTransformer(BaseAggregatorXApiTransformer):
+    """
+    Base transformer for completion aggregator progress events.
+    """
+
+    _verb = Verb(
+        id=constants.XAPI_VERB_PROGRESSED,
+        display=LanguageMap({constants.EN: constants.PROGRESSED}),
+    )
+    object_type = None
+    additional_fields = ('result', )
+
+    @openedx_filter(
+        filter_type="completion_aggregator.xapi.base_progress.get_object",
+    )
+    def get_object(self) -> Activity:
+        """
+        Get object for xAPI transformed event.
+        """
+        return super().get_object()
+
+    def get_result(self) -> Result:
+        """
+        Get result for xAPI transformed event.
+        """
+        return Result(
+            completion=self.get_data("data.percent") == 1.0,
+            score={
+                "scaled": self.get_data("data.percent") or 0
+            }
+        )
+
+
+@XApiTransformersRegistry.register("openedx.completion_aggregator.progress.chapter")
+@XApiTransformersRegistry.register("openedx.completion_aggregator.progress.sequential")
+@XApiTransformersRegistry.register("openedx.completion_aggregator.progress.vertical")
+class ModuleProgressTransformer(BaseProgressTransformer):
+    """
+    Transformer for event generated when a user makes progress in a section, subsection or unit.
+    """
+
+    object_type = constants.XAPI_ACTIVITY_MODULE
+
+
+@XApiTransformersRegistry.register("openedx.completion_aggregator.progress.course")
+class CourseProgressTransformer(BaseProgressTransformer):
+    """
+    Transformer for event generated when a user makes progress in a course.
+    """
+
+    object_type = constants.XAPI_ACTIVITY_COURSE
+
+
+class BaseCompletionTransformer(BaseAggregatorXApiTransformer):
+    """
+    Base transformer for aggregator completion events.
+    """
+
+    _verb = Verb(
+        id=constants.XAPI_VERB_COMPLETED,
+        display=LanguageMap({constants.EN: constants.COMPLETED}),
+    )
+    object_type = None
+
+    @openedx_filter(
+        filter_type="completion_aggregator.xapi.base_completion.get_object",
+    )
+    def get_object(self) -> Activity:
+        """
+        Get object for xAPI transformed event.
+        """
+        return super().get_object()
+
+
+@XApiTransformersRegistry.register("openedx.completion_aggregator.completion.chapter")
+@XApiTransformersRegistry.register("openedx.completion_aggregator.completion.sequential")
+@XApiTransformersRegistry.register("openedx.completion_aggregator.completion.vertical")
+class ModuleCompletionTransformer(BaseCompletionTransformer):
+    """
+    Transformer for events generated when a user completes a section, subsection or unit.
+    """
+
+    object_type = constants.XAPI_ACTIVITY_MODULE
+
+
+@XApiTransformersRegistry.register("openedx.completion_aggregator.completion.course")
+class CourseCompletionTransformer(BaseCompletionTransformer):
+    """
+    Transformer for event generated when a user completes a course.
+    """
+
+    object_type = constants.XAPI_ACTIVITY_COURSE

requirements/base.in

@@ -9,5 +9,7 @@ django-model-utils    # Provides TimeStampedModel abstract base class
 edx-opaque-keys    # Provides CourseKey and UsageKey
 edx-completion
 edx-toggles
+event-tracking
 six
 XBlock[django]
+edx-event-routing-backends  # Provides xAPI transforms for aggregator events