feat: load fake learner competency statuses

README.rst

@@ -330,6 +330,45 @@ test::
     # real enrollments dumped by the LMS, as with first_catalog_pathway_id.
     first_pathway_enrollment_id: 100001
 
+    # Fake organizations are named <org_prefix>0, <org_prefix>1... A different
+    # prefix keeps a load apart from earlier ones, or from real data, in the
+    # org filters.
+    org_prefix: Org
+
+    # The ids given to the first fake taxonomy, tag and object tag; the rest
+    # count up from them (tag and object tag ids across all taxonomies and
+    # courses). Raise them above the ids of real rows dumped by the LMS into
+    # the same event_sink tables, or the two overwrite each other.
+    first_taxonomy_id: 1
+    first_tag_id: 1
+    first_object_tag_id: 1
+
+    # The taxonomy export id written on every fake object tag. Aspects tells
+    # competency taxonomies apart by an export id starting with "competency-",
+    # so e.g. competency-music makes the fake taxonomy a competency taxonomy.
+    # Only then are learner competency statuses written to
+    # event_sink.student_competency_status, which exists only in Aspects installs
+    # that have run the tutor-contrib-aspects migration
+    # 0043_event_sink_student_competency_status. Each learner enrolled in a
+    # course gets a status on every competency tagged on its blocks: none,
+    # PartiallyAttempted or Demonstrated, with time to mastery spread from hours
+    # to weeks, and parent competencies rolled up the way openedx-core does.
+    taxonomy_export_id: fake export id
+
+    # The id given to the first fake Student Competency Status, kept above the
+    # ids of real statuses dumped by the LMS, as with first_catalog_pathway_id.
+    first_student_competency_status_id: 100001
+
+    # A competency load also writes each learner's status per course to
+    # event_sink.student_competency_criteria_group_status (tutor-contrib-aspects
+    # migration 0044_event_sink_student_competency_criteria_group_status): one
+    # course-level criteria group row per learner, competency and course, which
+    # can be AttemptedNotDemonstrated, the status core never stores per
+    # competency. The fake criteria groups exist only as ids. These are the first
+    # status id and the first group id, kept above the real ones as above.
+    first_competency_group_status_id: 100001
+    first_competency_criteria_group_id: 100001
+
     # The number of learners to create, random subsets of these will be
     # "registered" for each course and have statements generated for them
     # between their registration date and the end of the course

default_config.yaml

@@ -62,6 +62,32 @@ first_catalog_pathway_id: 100001
 num_enrollments_per_pathway: 5
 first_pathway_enrollment_id: 100001
 
+# Fake organizations are named <org_prefix>0, <org_prefix>1... A different
+# prefix keeps a load apart from earlier ones in the org filters.
+org_prefix: Org
+
+# Ids of the fake taxonomies, tags and object tags. Raise them above the ids of
+# real rows the LMS dumps into the same event_sink tables, or the two overwrite
+# each other.
+first_taxonomy_id: 1
+first_tag_id: 1
+first_object_tag_id: 1
+
+# Export id written on every fake object tag. One starting with "competency-"
+# (e.g. competency-music) makes the fake taxonomy a competency taxonomy and
+# generates learner competency statuses into event_sink.student_competency_status,
+# which needs the tutor-contrib-aspects migration
+# 0043_event_sink_student_competency_status.
+taxonomy_export_id: fake export id
+first_student_competency_status_id: 100001
+
+# Course-level criteria group statuses of a competency load, written to
+# event_sink.student_competency_criteria_group_status (tutor-contrib-aspects
+# migration 0044_event_sink_student_competency_criteria_group_status), and the ids
+# of the fake criteria groups they point at.
+first_competency_group_status_id: 100001
+first_competency_criteria_group_id: 100001
+
 # This replicates users updating their profiles several times, creating
 # more rows
 num_actor_profile_changes: 5

xapi_db_load/backends/chdb.py

@@ -85,6 +85,18 @@ def get_test_data_tasks(self) -> List[Waiter]:
                 self.event_generator,
                 self.db_loader,
             ),
+            WriteStudentCompetencyStatuses(
+                self.config,
+                self.logger,
+                self.event_generator,
+                self.db_loader,
+            ),
+            WriteStudentCompetencyCriteriaGroupStatuses(
+                self.config,
+                self.logger,
+                self.event_generator,
+                self.db_loader,
+            ),
             WriteBlocks(
                 self.config,
                 self.logger,
@@ -666,6 +678,124 @@ async def _run_task(self):
             await self._write_list_sql_retry(out_data, "pathway_enrollment")
 
 
+class WriteStudentCompetencyStatuses(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
+    """
+    Write the learners' competency statuses to S3 using CHDB, all under one dump.
+    """
+
+    task_name = "Insert Student Competency Statuses"
+    schema = "event_sink"
+    table = "student_competency_status"
+    s3_load_prefix = "student_competency_status"
+    table_definition = """
+    `id` Int64,
+    `actor_id` String,
+    `tag_id` Int64,
+    `taxonomy_id` Int64,
+    `status_id` UInt8,
+    `status` String,
+    `created` String,
+    `modified` String,
+    `dump_id` UUID,
+    `time_last_dumped` String
+    """
+
+    async def _run_task(self):
+        """
+        Insert the learners' competency statuses into the event sink db.
+        """
+        statuses = self.event_generator.student_competency_statuses
+        self.update_total_task_count(len(statuses))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = datetime.now(UTC)
+        out_data = []
+
+        for s in statuses:
+            out_data.append(
+                f"""(
+                {s["id"]},
+                '{s["actor_id"]}',
+                {s["tag_id"]},
+                {s["taxonomy_id"]},
+                {s["status_id"]},
+                '{s["status"]}',
+                '{s["created"]}',
+                '{s["modified"]}',
+                '{dump_id}',
+                '{dump_time}'
+            )"""
+            )
+            self.update_completed_task_count(increment_by=1)
+
+        if out_data:
+            await self._write_list_sql_retry(out_data, "student_competency_status")
+
+
+class WriteStudentCompetencyCriteriaGroupStatuses(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
+    """
+    Write the learners' course-level criteria group statuses to S3 using CHDB, all under one dump.
+    """
+
+    task_name = "Insert Student Competency Criteria Group Statuses"
+    schema = "event_sink"
+    table = "student_competency_criteria_group_status"
+    s3_load_prefix = "student_competency_criteria_group_status"
+    table_definition = """
+    `id` Int64,
+    `actor_id` String,
+    `group_id` Int64,
+    `parent_group_id` Nullable(Int64),
+    `tag_id` Int64,
+    `taxonomy_id` Int64,
+    `course_key` String,
+    `group_level` String,
+    `status_id` UInt8,
+    `status` String,
+    `created` String,
+    `modified` String,
+    `dump_id` UUID,
+    `time_last_dumped` String
+    """
+
+    async def _run_task(self):
+        """
+        Insert the learners' course-level criteria group statuses into the event sink db.
+        """
+        statuses = self.event_generator.student_competency_criteria_group_statuses
+        self.update_total_task_count(len(statuses))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = datetime.now(UTC)
+        out_data = []
+
+        for s in statuses:
+            out_data.append(
+                f"""(
+                {s["id"]},
+                '{s["actor_id"]}',
+                {s["group_id"]},
+                {s["parent_group_id"]},
+                {s["tag_id"]},
+                {s["taxonomy_id"]},
+                '{s["course_key"]}',
+                '{s["group_level"]}',
+                {s["status_id"]},
+                '{s["status"]}',
+                '{s["created"]}',
+                '{s["modified"]}',
+                '{dump_id}',
+                '{dump_time}'
+            )"""
+            )
+            self.update_completed_task_count(increment_by=1)
+
+        if out_data:
+            await self._write_list_sql_retry(
+                out_data, "student_competency_criteria_group_status"
+            )
+
+
 class WriteBlocks(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
     """
     Write the course block data to S3 using CHDB.
@@ -774,20 +904,17 @@ async def _run_task(self):
             obj_tag_out_data = []
             object_tags = course.serialize_object_tag_data_for_event_sink()
 
-            row_id = 0
             for _ in range(num_course_publishes):
                 dump_id = str(uuid.uuid4())
                 dump_time = datetime.now(UTC)
                 for obj_tag in object_tags:
-                    row_id += 1
-
                     out_tag = f"""(
-                    {row_id},
+                    {obj_tag["object_tag_id"]},
                     '{obj_tag["object_id"]}',
                     {obj_tag["taxonomy_id"]},
                     {obj_tag["tag_id"]},
                     '{obj_tag["value"]}',
-                    'fake export id',
+                    '{obj_tag["export_id"]}',
                     '{obj_tag["hierarchy"]}',
                     '{dump_id}',
                     '{dump_time}'
@@ -830,9 +957,9 @@ async def _run_task(self):
         dump_time = datetime.now(UTC)
         out_data = []
 
-        for taxonomy in taxonomies.keys():
+        for taxonomy, taxonomy_id in self.event_generator.taxonomy_ids.items():
             out = f"""(
-                {self.completed_task_count},
+                {taxonomy_id},
                 '{taxonomy}',
                 '{dump_id}',
                 '{dump_time}'

xapi_db_load/backends/clickhouse.py

@@ -45,6 +45,12 @@ def get_test_data_tasks(self) -> List[Waiter]:
             InsertCourses(self.config, self.logger, self.event_generator),
             InsertCatalogPathways(self.config, self.logger, self.event_generator),
             InsertPathwayEnrollments(self.config, self.logger, self.event_generator),
+            InsertStudentCompetencyStatuses(
+                self.config, self.logger, self.event_generator
+            ),
+            InsertStudentCompetencyCriteriaGroupStatuses(
+                self.config, self.logger, self.event_generator
+            ),
             InsertBlocks(self.config, self.logger, self.event_generator),
             InsertObjectTags(self.config, self.logger, self.event_generator),
             InsertTaxonomies(self.config, self.logger, self.event_generator),
@@ -351,6 +357,84 @@ async def _run_task(self):
             self.update_completed_task_count(increment_by=len(batch))
 
 
+class InsertStudentCompetencyStatuses(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
+    """Inserts Student Competency Status data to ClickHouse."""
+
+    task_name = "Insert Student Competency Statuses"
+
+    async def _run_task(self):
+        """
+        Insert the learners' competency statuses to ClickHouse, all under one dump, in batches.
+        """
+        statuses = self.event_generator.student_competency_statuses
+        self.update_total_task_count(len(statuses))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = str(datetime.now(UTC))
+        batch_size = self.config["batch_size"]
+
+        for i in range(0, len(statuses), batch_size):
+            batch = statuses[i:i + batch_size]
+            out_data = [
+                (
+                    s["id"],
+                    s["actor_id"],
+                    s["tag_id"],
+                    s["taxonomy_id"],
+                    s["status_id"],
+                    s["status"],
+                    s["created"],
+                    s["modified"],
+                    dump_id,
+                    dump_time,
+                )
+                for s in batch
+            ]
+            await self._insert_rows("student_competency_status", out_data)
+            self.update_completed_task_count(increment_by=len(batch))
+
+
+class InsertStudentCompetencyCriteriaGroupStatuses(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
+    """Inserts Student Competency Criteria Group Status data to ClickHouse."""
+
+    task_name = "Insert Student Competency Criteria Group Statuses"
+
+    async def _run_task(self):
+        """
+        Insert the learners' course-level criteria group statuses to ClickHouse, all under one dump, in batches.
+        """
+        statuses = self.event_generator.student_competency_criteria_group_statuses
+        self.update_total_task_count(len(statuses))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = str(datetime.now(UTC))
+        batch_size = self.config["batch_size"]
+
+        for i in range(0, len(statuses), batch_size):
+            batch = statuses[i:i + batch_size]
+            out_data = [
+                (
+                    s["id"],
+                    s["actor_id"],
+                    s["group_id"],
+                    s["parent_group_id"],
+                    s["tag_id"],
+                    s["taxonomy_id"],
+                    s["course_key"],
+                    s["group_level"],
+                    s["status_id"],
+                    s["status"],
+                    s["created"],
+                    s["modified"],
+                    dump_id,
+                    dump_time,
+                )
+                for s in batch
+            ]
+            await self._insert_rows("student_competency_criteria_group_status", out_data)
+            self.update_completed_task_count(increment_by=len(batch))
+
+
 class InsertBlocks(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
     """Inserts course block data to ClickHouse."""
 
@@ -418,17 +502,15 @@ async def _run_task(self):
                 dump_time = datetime.now(UTC)
                 obj_tag_out_data = []
 
-                row_id = 0
                 for obj_tag in object_tags:
-                    row_id += 1
                     obj_tag_out_data.append(
                         (
-                            row_id,
+                            obj_tag["object_tag_id"],
                             obj_tag["object_id"],
                             obj_tag["taxonomy_id"],
                             obj_tag["tag_id"],
                             obj_tag["value"],
-                            "fake export id",
+                            obj_tag["export_id"],
                             obj_tag["hierarchy"],
                             dump_id,
                             str(dump_time),
@@ -455,10 +537,8 @@ async def _run_task(self):
         self.update_total_task_count(len(taxonomies))
 
         out_data = []
-        tag_id = 0
-        for taxonomy in taxonomies.keys():
-            tag_id += 1
-            out_data.append((tag_id, taxonomy, dump_id, str(dump_time)))
+        for taxonomy, taxonomy_id in self.event_generator.taxonomy_ids.items():
+            out_data.append((taxonomy_id, taxonomy, dump_id, str(dump_time)))
             self.update_completed_task_count(increment_by=1)
 
         await self._insert_rows("taxonomy", out_data)