Load fake Catalog Pathways into event_sink.catalog_pathway

README.rst

@@ -308,6 +308,28 @@ test::
     # The number of organizations, courses will be evenly spread among these
     num_organizations: 3
 
+    # The number of Catalog Pathways written to event_sink.catalog_pathway,
+    # spread across the organizations and the Pathway Categories. That table
+    # exists only in Aspects installs that have run the tutor-contrib-aspects
+    # migration 0042_event_sink_pathways (released after v6.1.1). On older
+    # installs set this to 0, or the pathway insert fails.
+    num_catalog_pathways: 6
+
+    # The id given to the first fake Catalog Pathway; the rest count up from
+    # it. Keep it above the ids of real pathways dumped by the LMS into the
+    # same table so the two don't collide.
+    first_catalog_pathway_id: 100001
+
+    # The number of learners enrolled in each Catalog Pathway, written to
+    # event_sink.pathway_enrollment (same migration). Learners come from the
+    # pathway org's courses first, and are capped at num_actors. About 10% of
+    # the enrollments are unenrolled (is_active false).
+    num_enrollments_per_pathway: 5
+
+    # The id given to the first fake Pathway Enrollment, kept above the ids of
+    # real enrollments dumped by the LMS, as with first_catalog_pathway_id.
+    first_pathway_enrollment_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

@@ -50,6 +50,18 @@ course_length_days: 120
 num_organizations: 3
 num_actors: 10
 
+# Catalog Pathways (event_sink.catalog_pathway), spread across the organizations
+# and the Pathway Categories in constants.py. The table needs the
+# tutor-contrib-aspects migration 0042_event_sink_pathways (released after
+# v6.1.1); set num_catalog_pathways to 0 on older Aspects installs.
+num_catalog_pathways: 6
+first_catalog_pathway_id: 100001
+
+# Learners enrolled in each Catalog Pathway (event_sink.pathway_enrollment, same
+# migration), taken from the pathway org's courses first and capped at num_actors.
+num_enrollments_per_pathway: 5
+first_pathway_enrollment_id: 100001
+
 # This replicates users updating their profiles several times, creating
 # more rows
 num_actor_profile_changes: 5

xapi_db_load/backends/chdb.py

@@ -73,6 +73,18 @@ def get_test_data_tasks(self) -> List[Waiter]:
                 self.event_generator,
                 self.db_loader,
             ),
+            WriteCatalogPathways(
+                self.config,
+                self.logger,
+                self.event_generator,
+                self.db_loader,
+            ),
+            WritePathwayEnrollments(
+                self.config,
+                self.logger,
+                self.event_generator,
+                self.db_loader,
+            ),
             WriteBlocks(
                 self.config,
                 self.logger,
@@ -545,6 +557,115 @@ async def _run_task(self):
             self.update_completed_task_count(increment_by=1)
 
 
+class WriteCatalogPathways(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
+    """
+    Write the Catalog Pathways to S3 using CHDB, all under one dump.
+    """
+
+    task_name = "Insert Catalog Pathways"
+    schema = "event_sink"
+    table = "catalog_pathway"
+    s3_load_prefix = "catalog_pathway"
+    table_definition = """
+    `id` Int64,
+    `org` String,
+    `pathway_code` String,
+    `title` String,
+    `category_code` String,
+    `category_title` String,
+    `created` String,
+    `modified` String,
+    `dump_id` UUID,
+    `time_last_dumped` String,
+    `is_deleted` UInt8
+    """
+
+    async def _run_task(self):
+        """
+        Insert the Catalog Pathways into the event sink db.
+        """
+        pathways = self.event_generator.catalog_pathways
+        self.update_total_task_count(len(pathways))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = datetime.now(UTC)
+        out_data = []
+
+        for p in pathways:
+            # Titles such as "Master's Degree" need their quotes escaped for SQL.
+            title = p["title"].replace("'", "\\'")
+            category_title = p["category_title"].replace("'", "\\'")
+            out_data.append(
+                f"""(
+                {p["id"]},
+                '{p["org"]}',
+                '{p["pathway_code"]}',
+                '{title}',
+                '{p["category_code"]}',
+                '{category_title}',
+                '{p["created"]}',
+                '{p["modified"]}',
+                '{dump_id}',
+                '{dump_time}',
+                0
+            )"""
+            )
+            self.update_completed_task_count(increment_by=1)
+
+        if out_data:
+            await self._write_list_sql_retry(out_data, "catalog_pathway")
+
+
+class WritePathwayEnrollments(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
+    """
+    Write the Pathway Enrollments to S3 using CHDB, all under one dump.
+    """
+
+    task_name = "Insert Pathway Enrollments"
+    schema = "event_sink"
+    table = "pathway_enrollment"
+    s3_load_prefix = "pathway_enrollment"
+    table_definition = """
+    `id` Int64,
+    `catalog_pathway_id` Int64,
+    `actor_id` String,
+    `created` String,
+    `modified` String,
+    `is_active` Bool,
+    `dump_id` UUID,
+    `time_last_dumped` String
+    """
+
+    async def _run_task(self):
+        """
+        Insert the Pathway Enrollments into the event sink db.
+        """
+        enrollments = self.event_generator.pathway_enrollments
+        self.update_total_task_count(len(enrollments))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = datetime.now(UTC)
+        out_data = []
+
+        for e in enrollments:
+            out_data.append(
+                f"""(
+                {e["id"]},
+                {e["catalog_pathway_id"]},
+                '{e["actor_id"]}',
+                '{e["created"]}',
+                '{e["modified"]}',
+                {str(e["is_active"]).lower()},
+                '{dump_id}',
+                '{dump_time}'
+            )"""
+            )
+            self.update_completed_task_count(increment_by=1)
+
+        if out_data:
+            await self._write_list_sql_retry(out_data, "pathway_enrollment")
+
+
 class WriteBlocks(XAPILakeCHDBAsync):  # pylint: disable=abstract-method
     """
     Write the course block data to S3 using CHDB.

xapi_db_load/backends/clickhouse.py

@@ -43,6 +43,8 @@ def get_test_data_tasks(self) -> List[Waiter]:
             self.event_generator,
             InsertInitialEnrollments(self.config, self.logger, self.event_generator),
             InsertCourses(self.config, self.logger, self.event_generator),
+            InsertCatalogPathways(self.config, self.logger, self.event_generator),
+            InsertPathwayEnrollments(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),
@@ -278,6 +280,77 @@ async def _run_task(self):
             self.update_completed_task_count(increment_by=1)
 
 
+class InsertCatalogPathways(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
+    """Inserts Catalog Pathway data to ClickHouse."""
+
+    task_name = "Insert Catalog Pathways"
+
+    async def _run_task(self):
+        """
+        Insert the Catalog Pathways to ClickHouse, all under one dump.
+        """
+        pathways = self.event_generator.catalog_pathways
+        self.update_total_task_count(1)
+
+        dump_id = str(uuid.uuid4())
+        dump_time = datetime.now(UTC)
+        out_data = [
+            (
+                p["id"],
+                p["org"],
+                p["pathway_code"],
+                p["title"],
+                p["category_code"],
+                p["category_title"],
+                p["created"],
+                p["modified"],
+                dump_id,
+                str(dump_time),
+                0,  # is_deleted: fake pathways are never tombstones
+            )
+            for p in pathways
+        ]
+
+        if out_data:
+            await self._insert_rows("catalog_pathway", out_data)
+        self.update_completed_task_count(increment_by=1)
+
+
+class InsertPathwayEnrollments(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
+    """Inserts Pathway Enrollment data to ClickHouse."""
+
+    task_name = "Insert Pathway Enrollments"
+
+    async def _run_task(self):
+        """
+        Insert the Pathway Enrollments to ClickHouse, all under one dump, in batches.
+        """
+        enrollments = self.event_generator.pathway_enrollments
+        self.update_total_task_count(len(enrollments))
+
+        dump_id = str(uuid.uuid4())
+        dump_time = str(datetime.now(UTC))
+        batch_size = self.config["batch_size"]
+
+        for i in range(0, len(enrollments), batch_size):
+            batch = enrollments[i:i + batch_size]
+            out_data = [
+                (
+                    e["id"],
+                    e["catalog_pathway_id"],
+                    e["actor_id"],
+                    e["created"],
+                    e["modified"],
+                    e["is_active"],
+                    dump_id,
+                    dump_time,
+                )
+                for e in batch
+            ]
+            await self._insert_rows("pathway_enrollment", out_data)
+            self.update_completed_task_count(increment_by=len(batch))
+
+
 class InsertBlocks(XAPILakeClickhouseAsync):  # pylint: disable=abstract-method
     """Inserts course block data to ClickHouse."""
 

xapi_db_load/backends/csv.py

@@ -48,6 +48,8 @@ def get_test_data_tasks(self) -> List[Waiter]:
         return [
             self.event_generator,
             WriteCSVCourses(self.config, self.logger, self.event_generator),
+            WriteCSVCatalogPathways(self.config, self.logger, self.event_generator),
+            WriteCSVPathwayEnrollments(self.config, self.logger, self.event_generator),
             WriteCSVBlocks(self.config, self.logger, self.event_generator),
             WriteCSVObjectTags(self.config, self.logger, self.event_generator),
             WriteCSVTaxonomies(self.config, self.logger, self.event_generator),
@@ -224,6 +226,79 @@ async def _run_task(self):
                     self.update_completed_task_count(increment_by=1)
 
 
+class WriteCSVCatalogPathways(XAPILakeCSVAsync):
+    """Writes Catalog Pathway data to a gzipped CSV file."""
+
+    task_name = "Write CSV Catalog Pathways"
+    schema = "event_sink"
+    table = "catalog_pathway"
+    filename = "catalog_pathways.csv.gz"
+
+    async def _run_task(self):
+        """
+        Write out the Catalog Pathways data file, all under one dump.
+        """
+        async with self.get_csv_handle("catalog_pathways") as pathway_csv_writer:
+            pathways = self.event_generator.catalog_pathways
+            self.update_total_task_count(len(pathways))
+
+            dump_id = str(uuid.uuid4())
+            dump_time = datetime.now(UTC)
+
+            for p in pathways:
+                pathway_csv_writer.writerow(
+                    (
+                        p["id"],
+                        p["org"],
+                        p["pathway_code"],
+                        p["title"],
+                        p["category_code"],
+                        p["category_title"],
+                        p["created"],
+                        p["modified"],
+                        dump_id,
+                        dump_time,
+                        0,  # is_deleted: fake pathways are never tombstones
+                    )
+                )
+                self.update_completed_task_count(increment_by=1)
+
+
+class WriteCSVPathwayEnrollments(XAPILakeCSVAsync):
+    """Writes Pathway Enrollment data to a gzipped CSV file."""
+
+    task_name = "Write CSV Pathway Enrollments"
+    schema = "event_sink"
+    table = "pathway_enrollment"
+    filename = "pathway_enrollments.csv.gz"
+
+    async def _run_task(self):
+        """
+        Write out the Pathway Enrollments data file, all under one dump.
+        """
+        async with self.get_csv_handle("pathway_enrollments") as enrollment_csv_writer:
+            enrollments = self.event_generator.pathway_enrollments
+            self.update_total_task_count(len(enrollments))
+
+            dump_id = str(uuid.uuid4())
+            dump_time = datetime.now(UTC)
+
+            for e in enrollments:
+                enrollment_csv_writer.writerow(
+                    (
+                        e["id"],
+                        e["catalog_pathway_id"],
+                        e["actor_id"],
+                        e["created"],
+                        e["modified"],
+                        str(e["is_active"]).lower(),
+                        dump_id,
+                        dump_time,
+                    )
+                )
+                self.update_completed_task_count(increment_by=1)
+
+
 class WriteCSVBlocks(XAPILakeCSVAsync):
     """Writes course block data to a gzipped CSV file."""
 

xapi_db_load/backends/ralph.py

@@ -15,10 +15,12 @@
 from xapi_db_load.backends.base_async_backend import BaseBackendTasks
 from xapi_db_load.backends.clickhouse import (
     InsertBlocks,
+    InsertCatalogPathways,
     InsertCourses,
     InsertExternalIDs,
     InsertInitialEnrollments,
     InsertObjectTags,
+    InsertPathwayEnrollments,
     InsertProfiles,
     InsertTags,
     InsertTaxonomies,
@@ -41,6 +43,8 @@ def get_test_data_tasks(self):
             self.event_generator,
             InsertInitialEnrollments(self.config, self.logger, self.event_generator),
             InsertCourses(self.config, self.logger, self.event_generator),
+            InsertCatalogPathways(self.config, self.logger, self.event_generator),
+            InsertPathwayEnrollments(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),