Integrate canvas course export with google sheet (#1711)

* integrate canvas course export with google sheet

* add course_content_metadata to sensor

* update cursor

* updates

Author: rachellougee

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "polars ~= 1.19",
     "pyarrow ~=21.0.0",
     "pydantic ~=2.11.1",
+    "pygsheets>=2.0.6",
     "pymysql ~= 1.1",
     "s3fs (>=2025.2.0,<2026.0.0)",
     "trino[external-authentication-token-cache]>=0.336.0",

src/ol_orchestrate/assets/canvas.py

@@ -9,8 +9,8 @@
     AssetKey,
     AssetOut,
     DataVersion,
+    DynamicPartitionsDefinition,
     Output,
-    StaticPartitionsDefinition,
     asset,
     multi_asset,
 )
@@ -22,68 +22,7 @@
 )
 from ol_orchestrate.lib.utils import compute_zip_content_hash
 
-# predefined course IDs to export
-canvas_course_ids = StaticPartitionsDefinition(
-    [
-        "155",
-        "7023",
-        "14566",
-        "28766",
-        "28768",
-        "28770",
-        "28772",
-        "28774",
-        "28777",
-        "28751",
-        "28753",
-        "28755",
-        "28759",
-        "28765",
-        "28767",
-        "28785",
-        # "28760",
-        # "28761",
-        # "28762",
-        "28803",
-        "28805",
-        "28807",
-        "28808",
-        "28811",
-        "28813",
-        "28815",
-        "28816",
-        "28818",
-        "28821",
-        "28822",
-        "28824",
-        "28839",
-        "28841",
-        "28842",
-        "28845",
-        "28847",
-        "28849",
-        "28880",
-        "33162",
-        "33448",
-        "34062",
-        "34545",
-        "34562",
-        "34580",
-        "34593",
-        "34599",
-        "34625",
-        "34627",
-        "34631",
-        "34633",
-        "34642",
-        "34653",
-        "34681",
-        "34716",
-        "34721",
-        # "34859",
-        "35054",
-    ]
-)
+canvas_course_ids = DynamicPartitionsDefinition(name="canvas_course_ids")
 
 
 def _extract_course_files(context, course_id):

src/ol_orchestrate/definitions/canvas_course_export.py

@@ -1,7 +1,12 @@
+from typing import Any
+
 from dagster import (
+    ConfigurableResource,
     Definitions,
-    build_schedule_from_partitioned_job,
+    OpExecutionContext,
+    RunRequest,
     define_asset_job,
+    schedule,
 )
 from dagster_aws.s3 import S3Resource
 
@@ -15,24 +20,54 @@
     default_file_object_io_manager,
     default_io_manager,
 )
-from ol_orchestrate.lib.utils import authenticate_vault, s3_uploads_bucket
+from ol_orchestrate.lib.utils import (
+    authenticate_vault,
+    s3_uploads_bucket,
+)
 from ol_orchestrate.resources.api_client_factory import ApiClientFactory
+from ol_orchestrate.sensors.canvas import canvas_google_sheet_course_id_sensor
 
 vault = authenticate_vault(DAGSTER_ENV, VAULT_ADDRESS)
 
+gs_secrets = vault.client.secrets.kv.v1.read_secret(
+    mount_point="secret-data",
+    path="pipelines/google-service-account",
+)["data"]
+
+
+class GoogleSheetConfig(ConfigurableResource):
+    service_account_json: dict[str, Any]  # Service account JSON credentials
+    # Google Sheet ID for canvas course IDs
+    sheet_id: str = "13AoothEhEvWs2cJEEfZETm7E6h3-ZY4tD11KX_ARe1A"
+    worksheet_id: int = 1472315099  # Worksheet ID (gid) within the Google Sheet
 
-canvas_course_content_job = define_asset_job(
+
+# Asset job that will be executed per partition (course_id)
+canvas_course_export_job = define_asset_job(
     name="canvas_course_export_job",
     selection=[export_course_content, course_content_metadata],
     partitions_def=canvas_course_ids,
 )
 
-canvas_course_export_schedule = build_schedule_from_partitioned_job(
-    name="canvas_course_export_schedule",
-    job=canvas_course_content_job,
+
+@schedule(
     cron_schedule="0 */6 * * *",
+    job=canvas_course_export_job,
     execution_timezone="Etc/UTC",
+    required_resource_keys={"google_sheet_config"},
 )
+def canvas_course_export_schedule(context: OpExecutionContext):
+    """Return a RunRequest for each canvas course ID found in the Google Sheet"""
+    partition_keys = context.instance.get_dynamic_partitions("canvas_course_ids")
+
+    return [
+        RunRequest(
+            run_key=partition_key,
+            partition_key=partition_key,
+        )
+        for partition_key in partition_keys
+    ]
+
 
 canvas_course_export = Definitions(
     resources={
@@ -60,7 +95,9 @@
             kv_version="2",
             vault=vault,
         ),
+        "google_sheet_config": GoogleSheetConfig(service_account_json=gs_secrets),
     },
     assets=[export_course_content, course_content_metadata],
     schedules=[canvas_course_export_schedule],
+    sensors=[canvas_google_sheet_course_id_sensor],
 )

src/ol_orchestrate/lib/canvas.py

@@ -0,0 +1,54 @@
+import json
+
+import pygsheets
+from dagster import OpExecutionContext
+from google.oauth2 import service_account
+
+
+def fetch_canvas_course_ids_from_google_sheet(context: OpExecutionContext):
+    """
+    Fetch all canvas course IDs from a Google Sheet
+    """
+    sheet_config = context.resources.google_sheet_config
+
+    if sheet_config.service_account_json is None:
+        context.log.error("No google service account credentials found in vault")
+        return set()
+
+    creds_dict = (
+        sheet_config.service_account_json
+        if isinstance(sheet_config.service_account_json, dict)
+        else json.loads(sheet_config.service_account_json)
+    )
+
+    creds_dict["private_key"] = creds_dict["private_key"].replace("\\n", "\n")
+
+    scopes = [
+        "https://www.googleapis.com/auth/spreadsheets",
+        "https://www.googleapis.com/auth/drive",
+    ]
+
+    credentials = service_account.Credentials.from_service_account_info(
+        creds_dict, scopes=scopes
+    )
+
+    google_sheet_client = pygsheets.authorize(custom_credentials=credentials)
+    spreadsheet = google_sheet_client.open_by_key(sheet_config.sheet_id)
+
+    # Find worksheet by gid
+    worksheet = next(
+        (
+            worksheet
+            for worksheet in spreadsheet.worksheets()
+            if worksheet.id == sheet_config.worksheet_id
+        ),
+        None,
+    )
+    if worksheet is None:
+        context.log.error("No worksheet found with gid %s", sheet_config.worksheet_id)
+        return set()
+
+    # Get all values from the first column and filter to only numeric values
+    column_values = worksheet.get_col(1, include_tailing_empty=False)
+
+    return {value.strip() for value in column_values if value.strip().isdigit()}

src/ol_orchestrate/sensors/canvas.py

@@ -0,0 +1,79 @@
+import json
+
+from dagster import (
+    AddDynamicPartitionsRequest,
+    AssetKey,
+    DefaultSensorStatus,
+    DeleteDynamicPartitionsRequest,
+    RunRequest,
+    SensorResult,
+    SkipReason,
+    sensor,
+)
+
+from ol_orchestrate.lib.canvas import fetch_canvas_course_ids_from_google_sheet
+
+
+@sensor(
+    description="Sensor to monitor a Google Sheet for Canvas course IDs to export.",
+    minimum_interval_seconds=60 * 60,  # Check every 1 hour
+    required_resource_keys={"google_sheet_config"},
+    default_status=DefaultSensorStatus.STOPPED,
+    asset_selection=[
+        AssetKey(["canvas", "course_content"]),
+        AssetKey(["canvas", "course_metadata"]),
+        AssetKey(["canvas", "course_content_metadata"]),
+    ],
+)
+def canvas_google_sheet_course_id_sensor(context):
+    google_sheet_course_ids = fetch_canvas_course_ids_from_google_sheet(context)
+    context.log.info("google_sheet_course_ids: %s", google_sheet_course_ids)
+
+    # Existing dynamic partitions
+    existing_partitions = set(
+        context.instance.get_dynamic_partitions("canvas_course_ids")
+    )
+    context.log.info("existing_partitions: %s", existing_partitions)
+
+    # Register any new course IDs as partitions
+    new_course_ids = google_sheet_course_ids - existing_partitions
+    removed_course_ids = existing_partitions - google_sheet_course_ids
+
+    if not new_course_ids and not removed_course_ids:
+        return SkipReason("No changes in canvas course IDs")
+
+    run_requests = [
+        RunRequest(
+            asset_selection=[
+                AssetKey(["canvas", "course_content"]),
+                AssetKey(["canvas", "course_metadata"]),
+                AssetKey(["canvas", "course_content_metadata"]),
+            ],
+            partition_key=course_id,
+        )
+        for course_id in new_course_ids
+    ]
+
+    updated_ids = sorted(google_sheet_course_ids)
+    dynamic_requests = []
+    if new_course_ids:
+        dynamic_requests.append(
+            AddDynamicPartitionsRequest(
+                partitions_def_name="canvas_course_ids",
+                partition_keys=list(new_course_ids),
+            )
+        )
+
+    if removed_course_ids:
+        dynamic_requests.append(
+            DeleteDynamicPartitionsRequest(
+                partitions_def_name="canvas_course_ids",
+                partition_keys=list(removed_course_ids),
+            )
+        )
+
+    return SensorResult(
+        dynamic_partitions_requests=dynamic_requests,
+        run_requests=run_requests,
+        cursor=json.dumps(updated_ids),
+    )