diff --git a/README.rst b/README.rst index 4b451966..2fbc2430 100644 --- a/README.rst +++ b/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. diff --git a/src/platform_plugin_aspects/settings/common.py b/src/platform_plugin_aspects/settings/common.py index fbd991bd..37ba256a 100644 --- a/src/platform_plugin_aspects/settings/common.py +++ b/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 diff --git a/src/platform_plugin_aspects/signals.py b/src/platform_plugin_aspects/signals.py index 5d1ef689..6fbe06f6 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/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) diff --git a/src/platform_plugin_aspects/sinks/__init__.py b/src/platform_plugin_aspects/sinks/__init__.py index f3c059a6..7f9f283f 100644 --- a/src/platform_plugin_aspects/sinks/__init__.py +++ b/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 diff --git a/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py new file mode 100644 index 00000000..dccf6061 --- /dev/null +++ b/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") diff --git a/src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py b/src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py new file mode 100644 index 00000000..80c1250c --- /dev/null +++ b/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") diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index b20315d1..73caa0fa 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/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): @@ -210,6 +214,71 @@ def get_course_key(self, obj): 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.""" diff --git a/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py new file mode 100644 index 00000000..8a13a147 --- /dev/null +++ b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py @@ -0,0 +1,106 @@ +""" +Test the catalog pathway sink module. +""" + +import logging +from unittest.mock import Mock, patch + +import pytest +import requests +import responses +from responses.registries import OrderedRegistry + +from platform_plugin_aspects.sinks import ( + CatalogPathwaySink, + CatalogPathwayTombstoneSink, +) + +log = logging.getLogger(__name__) + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_get_queryset(mock_get_queryset): + """ + Test the get_queryset method. + """ + sink = CatalogPathwaySink(None, None) + + sink.get_queryset() + + mock_get_queryset.assert_called_once_with(None) + mock_get_queryset.return_value.select_related.assert_called_once_with( + "org", "category" + ) + + +@responses.activate( # pylint: disable=unexpected-keyword-arg,no-value-for-parameter + registry=OrderedRegistry +) +def test_tombstone_dump(): + """ + Test that the tombstone sink copies the pathway's latest row with is_deleted set. + """ + tombstone = responses.post( + "https://foo.bar/", + match=[ + responses.matchers.query_param_matcher( + { + "query": "INSERT INTO cool_data.catalog_pathway " + "SELECT * REPLACE (1 AS is_deleted, generateUUIDv4() AS dump_id, " + "now64(6) AS time_last_dumped) " + "FROM (SELECT * FROM cool_data.catalog_pathway WHERE id = 7 " + "ORDER BY time_last_dumped DESC LIMIT 1)", + } + ) + ], + headers={"X-ClickHouse-Summary": '{"written_rows":"1"}'}, + ) + mock_log = Mock() + + CatalogPathwayTombstoneSink(None, mock_log).dump("7") + + assert tombstone.call_count == 1 + mock_log.warning.assert_not_called() + + +@responses.activate( # pylint: disable=unexpected-keyword-arg,no-value-for-parameter + registry=OrderedRegistry +) +def test_tombstone_dump_retries_once(): + """ + Test that a failed tombstone is retried once, and a second failure raises. + """ + responses.post("https://foo.bar/", body=requests.ConnectionError()) + responses.post("https://foo.bar/") + + CatalogPathwayTombstoneSink(None, log).dump("7") + + responses.post("https://foo.bar/", body=requests.ConnectionError()) + responses.post("https://foo.bar/", body=requests.ConnectionError()) + + with pytest.raises(requests.ConnectionError): + CatalogPathwayTombstoneSink(None, log).dump("7") + + +@responses.activate # pylint: disable=unexpected-keyword-arg,no-value-for-parameter +def test_tombstone_dump_warns_when_nothing_written(): + """ + Test that a pathway that was never dumped logs a warning. + """ + responses.post( + "https://foo.bar/", + headers={"X-ClickHouse-Summary": '{"written_rows":"0"}'}, + ) + mock_log = Mock() + + CatalogPathwayTombstoneSink(None, mock_log).dump("7") + + mock_log.warning.assert_called_once() + + +def test_tombstone_dump_rejects_non_integer_id(): + """ + Test that an id that isn't an integer never reaches the query. + """ + with pytest.raises(ValueError): + CatalogPathwayTombstoneSink(None, log).dump("1 OR 1=1") diff --git a/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py b/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py new file mode 100644 index 00000000..a597be4f --- /dev/null +++ b/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py @@ -0,0 +1,20 @@ +""" +Test the pathway enrollment sink module. +""" + +from unittest.mock import patch + +from platform_plugin_aspects.sinks import PathwayEnrollmentSink + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_get_queryset(mock_get_queryset): + """ + Test the get_queryset method. + """ + sink = PathwayEnrollmentSink(None, None) + + sink.get_queryset() + + mock_get_queryset.assert_called_once_with(None) + mock_get_queryset.return_value.select_related.assert_called_once_with("user") diff --git a/src/platform_plugin_aspects/sinks/tests/test_serializers.py b/src/platform_plugin_aspects/sinks/tests/test_serializers.py index 8e281e86..c34406a7 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_serializers.py +++ b/src/platform_plugin_aspects/sinks/tests/test_serializers.py @@ -6,8 +6,10 @@ from platform_plugin_aspects.sinks.serializers import ( BaseSinkSerializer, + CatalogPathwaySerializer, CourseOverviewSerializer, DateTimeJSONEncoder, + PathwayEnrollmentSerializer, ) from test_utils.helpers import course_key_factory @@ -94,3 +96,39 @@ def test_get_course_key(self): mock_id = Mock() mock_overview = Mock(id=mock_id) self.assertEqual(self.serializer.get_course_key(mock_overview), str(mock_id)) + + +class TestCatalogPathwaySerializer(TestCase): + """ + Test CatalogPathwaySerializer + """ + + def test_is_deleted(self): + """ + Test that saved pathways are written with is_deleted 0, in the last column. + """ + serializer = CatalogPathwaySerializer() + + self.assertEqual(serializer.get_is_deleted(Mock()), 0) + self.assertEqual(CatalogPathwaySerializer.Meta.fields[-1], "is_deleted") + + +class TestPathwayEnrollmentSerializer(TestCase): + """ + Test PathwayEnrollmentSerializer + """ + + @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" + ) + enrollment = Mock() + + actor_id = PathwayEnrollmentSerializer().get_actor_id(enrollment) + + assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18" + mock_get_or_create_xapi_actor_id.assert_called_once_with(enrollment.user) diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 5e1fe66f..9ef61b77 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -2,15 +2,19 @@ Tests for signal handlers. """ -from unittest.mock import Mock, patch +from unittest.mock import ANY, Mock, patch from django.test import TestCase from platform_plugin_aspects.signals import ( + on_catalog_pathway_deleted, + on_catalog_pathway_saved_txn, on_externalid_saved_txn, + on_pathway_enrollment_saved_txn, on_user_retirement, receive_course_publish, ) +from platform_plugin_aspects.sinks import CatalogPathwayTombstoneSink from platform_plugin_aspects.sinks.user_retire_sink import UserRetirementSink @@ -41,6 +45,49 @@ def test_on_externalid_saved(self, mock_transaction): mock_transaction.on_commit.assert_called_once() + @patch("platform_plugin_aspects.signals.transaction") + def test_on_catalog_pathway_saved(self, mock_transaction): + """ + Test that on_catalog_pathway_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_catalog_pathway_saved_txn(sender, instance) + + mock_transaction.on_commit.assert_called_once() + + @patch("platform_plugin_aspects.signals.transaction") + def test_on_pathway_enrollment_saved(self, mock_transaction): + """ + Test that on_pathway_enrollment_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_pathway_enrollment_saved_txn(sender, instance) + + mock_transaction.on_commit.assert_called_once() + + @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") + @patch("platform_plugin_aspects.signals.transaction") + def test_on_catalog_pathway_deleted(self, mock_transaction, mock_dump_task): + """ + Test that on_catalog_pathway_deleted queues the tombstone sink on commit, with the id + read before Django clears it. + """ + instance = Mock(id=7) + on_catalog_pathway_deleted(Mock(), instance) + instance.id = None + + mock_dump_task.delay.assert_not_called() + mock_transaction.on_commit.assert_called_once_with(ANY, robust=True) + mock_transaction.on_commit.call_args.args[0]() + + mock_dump_task.delay.assert_called_once_with( + sink_module=CatalogPathwayTombstoneSink.__module__, + sink_name="CatalogPathwayTombstoneSink", + object_id="7", + ) + @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") def test_on_user_retirement(self, mock_dump_task): """ diff --git a/src/platform_plugin_aspects/tests/test_utils.py b/src/platform_plugin_aspects/tests/test_utils.py index 97fc6df8..bed8e620 100644 --- a/src/platform_plugin_aspects/tests/test_utils.py +++ b/src/platform_plugin_aspects/tests/test_utils.py @@ -14,6 +14,7 @@ generate_superset_context, get_ccx_courses, get_model, + get_or_create_xapi_actor_id, get_tags_for_block, get_user_dashboard_locale, ) @@ -292,3 +293,34 @@ def test_get_user_dashboard_locale(self, mock_get_model): user = Mock() assert get_user_dashboard_locale(user) == "en" mock_get_model.assert_called_once() + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_or_create_xapi_actor_id(self, mock_get_model): + """Test that get_or_create_xapi_actor_id returns the user's xapi ExternalId.""" + external_id = Mock(external_user_id="29e5b22b-b457-49fa-9633-6c0123305d18") + mock_get_model.return_value.add_new_user_id.return_value = (external_id, True) + user = Mock() + + assert ( + get_or_create_xapi_actor_id(user) == "29e5b22b-b457-49fa-9633-6c0123305d18" + ) + mock_get_model.assert_called_once_with("external_id") + mock_get_model.return_value.add_new_user_id.assert_called_once_with( + user, "xapi" + ) + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_or_create_xapi_actor_id_missing_type(self, mock_get_model): + """Test that get_or_create_xapi_actor_id fails loudly without the xapi ExternalIdType.""" + mock_get_model.return_value.add_new_user_id.return_value = (None, False) + + with self.assertRaises(ValueError): + get_or_create_xapi_actor_id(Mock()) + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_or_create_xapi_actor_id_missing_model(self, mock_get_model): + """Test that get_or_create_xapi_actor_id fails loudly without the ExternalId model.""" + mock_get_model.return_value = None + + with self.assertRaises(ValueError): + get_or_create_xapi_actor_id(Mock()) diff --git a/src/platform_plugin_aspects/utils.py b/src/platform_plugin_aspects/utils.py index 272c5e29..0783abe6 100644 --- a/src/platform_plugin_aspects/utils.py +++ b/src/platform_plugin_aspects/utils.py @@ -39,6 +39,9 @@ def _(text): DEFAULT_USER_LANGUAGE = "en" +# ExternalIdType name event-routing-backends uses for the xAPI actor id. +XAPI_EXTERNAL_ID_TYPE = "xapi" + def generate_superset_context( context, @@ -311,6 +314,28 @@ def get_tags_for_block(usage_key) -> set: return list(serialized_tags) +def get_or_create_xapi_actor_id(user): + """ + Return the learner's xAPI actor id, creating it if the learner has none yet. + + This is the id event-routing-backends sends as the xAPI actor (the + learner's ExternalId of type "xapi"), so sink rows keyed by it join the + xAPI tables and hold no PII. Creating the ExternalId fires its post_save, + which queues an ExternalIdSink dump. + + Raises ValueError if the ExternalId model or the "xapi" ExternalIdType is missing. + """ + ExternalId = get_model("external_id") + if not ExternalId: + raise ValueError("The ExternalId model could not be loaded.") + external_id, _ = ExternalId.add_new_user_id(user, XAPI_EXTERNAL_ID_TYPE) + if not external_id: + raise ValueError( + f"The '{XAPI_EXTERNAL_ID_TYPE}' ExternalIdType does not exist." + ) + return str(external_id.external_user_id) + + def get_user_dashboard_locale(user): """