mitodl
diff --git a/‎dg_projects/learning_resources/learning_resources/assets/mit_climate.py‎
Lines changed: 119 additions & 0 deletions b/‎dg_projects/learning_resources/learning_resources/assets/mit_climate.py‎
Lines changed: 119 additions & 0 deletions
diff --git a/‎dg_projects/learning_resources/learning_resources/assets/mit_edx_programs.py‎
Lines changed: 141 additions & 0 deletions b/‎dg_projects/learning_resources/learning_resources/assets/mit_edx_programs.py‎
Lines changed: 141 additions & 0 deletions
diff --git a/‎dg_projects/learning_resources/learning_resources/assets/mitpe.py‎
Lines changed: 115 additions & 0 deletions b/‎dg_projects/learning_resources/learning_resources/assets/mitpe.py‎
Lines changed: 115 additions & 0 deletions
@@ -0,0 +1,119 @@
+"""
+MIT Climate Portal webhook delivery asset.
+
+Reads pre-transformed article records from the
+``integrations__learn__mit_climate_articles`` Iceberg table (produced by dbt
+from the mit_climate dlt pipeline) and delivers them to MIT Learn via a
+single signed webhook POST.
+
+Data flow:
+    raw__mit_climate__api__articles (Iceberg, via dlt)
+        → integrations__learn__mit_climate_articles (dbt)
+            → MIT Learn webhook (this asset)
+
+Scheduling: daily at 06:00 UTC. Configured in definitions.py.
+"""
+
+import logging
+from typing import Any, cast
+
+import httpx2 as httpx
+import polars as pl
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    MetadataValue,
+    RetryPolicy,
+    asset,
+)
+from ol_orchestrate.lib.constants import DAGSTER_ENV
+from ol_orchestrate.lib.glue_helper import get_dbt_model_as_dataframe
+from ol_orchestrate.resources.api_client_factory import ApiClientFactory
+from ol_orchestrate.resources.learn_api import MITLearnApiClient
+
+log = logging.getLogger(__name__)
+
+_GLUE_DB = (
+    f"ol_warehouse_{DAGSTER_ENV}_integrations"
+    if DAGSTER_ENV in ("qa", "production")
+    else "ol_warehouse_production_integrations"
+)
+_TABLE = "integrations__learn__mit_climate_articles"
+
+
+def _row_to_resource(row: dict[str, Any]) -> dict[str, Any]:
+    """Map an integrations table row to the MIT Learn LearningResource shape."""
+    image_url = row.get("image_url")
+    image_alt = row.get("image_alt") or ""
+    topics_raw = row.get("topics") or ""
+    topics = [{"name": t.strip()} for t in topics_raw.split(",") if t.strip()]
+
+    return {
+        "readable_id": row["readable_id"],
+        "title": row["title"],
+        "url": row.get("url"),
+        "description": row.get("description"),
+        "image": {"url": image_url, "alt": image_alt} if image_url else None,
+        "topics": topics,
+        "published": True,
+        "etl_source": "mit_climate",
+        "offered_by": {"code": "climate"},
+        "platform": "climate",
+        # MIT Learn has no "article" resource_type; the legacy Celery ETL loads
+        # Climate articles as documents (load_documents) with an "Article"
+        # resource_category, so match that contract for the webhook path too.
+        "resource_type": "document",
+        "resource_category": "Article",
+        "created_on": row.get("last_modified"),
+    }
+
+
+@asset(
+    key=AssetKey(["mit_learn_delivery", "mit_climate_webhook"]),
+    group_name="mit_learn_delivery",
+    description=(
+        "Read MIT Climate Portal articles from the "
+        "integrations__learn__mit_climate_articles "
+        "Iceberg table and POST as a signed webhook batch to MIT Learn."
+    ),
+    deps=[
+        AssetKey(["integrations", "learn", "integrations__learn__mit_climate_articles"])
+    ],
+    retry_policy=RetryPolicy(max_retries=3, delay=5.0),
+)
+def mit_climate_webhook(
+    context: AssetExecutionContext,
+    learn_api: ApiClientFactory,
+) -> dict[str, Any]:
+    """Deliver MIT Climate articles to MIT Learn via signed webhook."""
+    context.log.info("Reading %s from Glue database %s", _TABLE, _GLUE_DB)
+    df: pl.DataFrame = get_dbt_model_as_dataframe(
+        database_name=_GLUE_DB,
+        table_name=_TABLE,
+    ).collect()
+    context.log.info("Loaded %d MIT Climate articles from Iceberg", len(df))
+
+    resources = [_row_to_resource(row) for row in df.iter_rows(named=True)]
+
+    context.log.info(
+        "Delivering %d MIT Climate articles to MIT Learn webhook", len(resources)
+    )
+    try:
+        response = cast(MITLearnApiClient, learn_api.client).notify_learning_resources(
+            resources
+        )
+    except httpx.HTTPStatusError as exc:
+        msg = (
+            f"MIT Climate webhook failed with status {exc.response.status_code}: {exc}"
+        )
+        context.log.exception(msg)
+        raise RuntimeError(msg) from exc
+
+    context.add_output_metadata(
+        {
+            "resource_count": len(resources),
+            "webhook_status": "success",
+            "response": MetadataValue.json(response),
+        }
+    )
+    return {"resource_count": len(resources), "webhook_status": "success"}
@@ -0,0 +1,141 @@
+"""
+MIT edX (MITx on edX.org) programs webhook delivery asset.
+
+Reads pre-transformed program records from the
+``integrations__learn__mit_edx_programs`` Iceberg table (produced by dbt from
+the mit_edx_programs dlt pipeline) and delivers them to MIT Learn via a single
+signed webhook POST.
+
+MicroMasters programs are excluded — the dlt pipeline filters them at ingest
+time and they are handled separately by the Cohort 1 Trino-pull path via
+integrations__learn__micromasters_programs.
+
+Data flow:
+    raw__edxorg__discovery__api__programs (Iceberg, via dlt)
+        → integrations__learn__mit_edx_programs (dbt)
+            → MIT Learn webhook (this asset)
+
+Scheduling: daily at 06:45 UTC. Configured in definitions.py.
+"""
+
+import json
+import logging
+from typing import Any, cast
+
+import httpx2 as httpx
+import polars as pl
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    MetadataValue,
+    RetryPolicy,
+    asset,
+)
+from ol_orchestrate.lib.constants import DAGSTER_ENV
+from ol_orchestrate.lib.glue_helper import get_dbt_model_as_dataframe
+from ol_orchestrate.resources.api_client_factory import ApiClientFactory
+from ol_orchestrate.resources.learn_api import MITLearnApiClient
+
+log = logging.getLogger(__name__)
+
+_GLUE_DB = (
+    f"ol_warehouse_{DAGSTER_ENV}_integrations"
+    if DAGSTER_ENV in ("qa", "production")
+    else "ol_warehouse_production_integrations"
+)
+_TABLE = "integrations__learn__mit_edx_programs"
+
+
+def _parse_topics(topics_json: str | None) -> list[dict[str, str]]:
+    """Parse the subjects JSON array into MIT Learn topic dicts."""
+    if not topics_json:
+        return []
+    try:
+        subjects = json.loads(topics_json)
+        return [{"name": s["name"]} for s in subjects if s.get("name")]
+    except (json.JSONDecodeError, TypeError):
+        log.warning("Could not parse topics_json: %r", topics_json)
+        return []
+
+
+def _parse_courses(courses_json: str | None) -> list[dict[str, str]]:
+    """Parse the courses JSON array into MIT Learn course readable_id dicts."""
+    if not courses_json:
+        return []
+    try:
+        courses = json.loads(courses_json)
+        return [{"readable_id": c["key"]} for c in courses if c.get("key")]
+    except (json.JSONDecodeError, TypeError):
+        log.warning("Could not parse courses_json: %r", courses_json)
+        return []
+
+
+def _row_to_resource(row: dict[str, Any]) -> dict[str, Any]:
+    """Map an integrations table row to the MIT Learn LearningResource shape."""
+    image_url = row.get("image_url")
+
+    return {
+        "readable_id": row["readable_id"],
+        "title": row["title"],
+        "url": row.get("url"),
+        "description": row.get("description"),
+        "image": {"url": image_url, "alt": row.get("title", "")} if image_url else None,
+        "topics": _parse_topics(row.get("topics_json")),
+        "published": True,
+        "etl_source": "mit_edx",
+        "offered_by": {"code": "mitx"},
+        "platform": "edxorg",
+        "resource_type": "program",
+        "courses": _parse_courses(row.get("courses_json")),
+    }
+
+
+@asset(
+    key=AssetKey(["mit_learn_delivery", "mit_edx_programs_webhook"]),
+    group_name="mit_learn_delivery",
+    description=(
+        "Read active MIT-authored edX.org programs from the "
+        "integrations__learn__mit_edx_programs Iceberg table and POST as a "
+        "signed webhook batch to MIT Learn. Excludes MicroMasters (handled via "
+        "the Cohort 1 Trino-pull path)."
+    ),
+    deps=[AssetKey(["integrations", "learn", "integrations__learn__mit_edx_programs"])],
+    retry_policy=RetryPolicy(max_retries=3, delay=10.0),
+)
+def mit_edx_programs_webhook(
+    context: AssetExecutionContext,
+    learn_api: ApiClientFactory,
+) -> dict[str, Any]:
+    """Deliver MIT edX programs to MIT Learn via signed webhook."""
+    context.log.info("Reading %s from Glue database %s", _TABLE, _GLUE_DB)
+    df: pl.DataFrame = get_dbt_model_as_dataframe(
+        database_name=_GLUE_DB,
+        table_name=_TABLE,
+    ).collect()
+    context.log.info("Loaded %d MIT edX programs from Iceberg", len(df))
+
+    resources = [_row_to_resource(row) for row in df.iter_rows(named=True)]
+
+    context.log.info(
+        "Delivering %d MIT edX programs to MIT Learn webhook", len(resources)
+    )
+    try:
+        response = cast(MITLearnApiClient, learn_api.client).notify_learning_resources(
+            resources
+        )
+    except httpx.HTTPStatusError as exc:
+        msg = (
+            f"MIT edX programs webhook failed with status "
+            f"{exc.response.status_code}: {exc}"
+        )
+        context.log.exception(msg)
+        raise RuntimeError(msg) from exc
+
+    context.add_output_metadata(
+        {
+            "delivered_count": len(resources),
+            "webhook_status": "success",
+            "response": MetadataValue.json(response),
+        }
+    )
+    return {"delivered_count": len(resources), "webhook_status": "success"}
@@ -0,0 +1,115 @@
+"""
+MIT Professional Education (MIT PE) webhook delivery asset.
+
+Reads pre-transformed course records from the
+``integrations__learn__mitpe_courses`` Iceberg table (produced by dbt from the
+mitpe dlt pipeline) and delivers them to MIT Learn via a signed webhook POST.
+
+Note: MIT PE does not expose a separate programs API endpoint. All records
+from the /feeds/courses/ feed are delivered as courses (resource_type='course').
+
+Data flow:
+    raw__mitpe__api__courses (Iceberg, via dlt)
+        → integrations__learn__mitpe_courses (dbt)
+            → MIT Learn webhook (this asset)
+
+Scheduling: daily at 06:15 UTC. Configured in definitions.py.
+"""
+
+import logging
+from typing import Any, cast
+
+import httpx2 as httpx
+import polars as pl
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    MetadataValue,
+    RetryPolicy,
+    asset,
+)
+from ol_orchestrate.lib.constants import DAGSTER_ENV
+from ol_orchestrate.lib.glue_helper import get_dbt_model_as_dataframe
+from ol_orchestrate.resources.api_client_factory import ApiClientFactory
+from ol_orchestrate.resources.learn_api import MITLearnApiClient
+
+log = logging.getLogger(__name__)
+
+_GLUE_DB = (
+    f"ol_warehouse_{DAGSTER_ENV}_integrations"
+    if DAGSTER_ENV in ("qa", "production")
+    else "ol_warehouse_production_integrations"
+)
+_COURSES_TABLE = "integrations__learn__mitpe_courses"
+
+
+def _row_to_resource(row: dict[str, Any]) -> dict[str, Any]:
+    """Map an integrations table row to the MIT Learn LearningResource shape."""
+    image_url = row.get("image_url")
+    image_alt = row.get("image_alt") or ""
+    topics_raw = row.get("topics") or ""
+    topics = [{"name": t.strip()} for t in topics_raw.split(",") if t.strip()]
+
+    return {
+        "readable_id": row["readable_id"],
+        "title": row["title"],
+        "url": row.get("url"),
+        "description": row.get("description"),
+        "image": {"url": image_url, "alt": image_alt} if image_url else None,
+        "topics": topics,
+        "published": True,
+        "professional": True,
+        "etl_source": "mitpe",
+        "offered_by": {"code": "mitpe"},
+        "platform": "mitpe",
+        "resource_type": row.get("resource_type", "course"),
+    }
+
+
+@asset(
+    key=AssetKey(["mit_learn_delivery", "mitpe_webhook"]),
+    group_name="mit_learn_delivery",
+    description=(
+        "Read MIT Professional Education courses from the "
+        "integrations__learn__mitpe_courses Iceberg table and POST as a signed "
+        "webhook batch to MIT Learn."
+    ),
+    deps=[
+        AssetKey(["integrations", "learn", "integrations__learn__mitpe_courses"]),
+    ],
+    retry_policy=RetryPolicy(max_retries=3, delay=5.0),
+)
+def mitpe_webhook(
+    context: AssetExecutionContext,
+    learn_api: ApiClientFactory,
+) -> dict[str, Any]:
+    """Deliver MIT PE courses to MIT Learn via signed webhook."""
+    context.log.info("Reading %s from Glue database %s", _COURSES_TABLE, _GLUE_DB)
+
+    courses_df: pl.DataFrame = get_dbt_model_as_dataframe(
+        database_name=_GLUE_DB, table_name=_COURSES_TABLE
+    ).collect()
+    context.log.info("Loaded %d MIT PE courses from Iceberg", len(courses_df))
+
+    resources = [_row_to_resource(row) for row in courses_df.iter_rows(named=True)]
+
+    context.log.info(
+        "Delivering %d MIT PE courses to MIT Learn webhook", len(resources)
+    )
+    try:
+        response = cast(MITLearnApiClient, learn_api.client).notify_learning_resources(
+            resources
+        )
+    except httpx.HTTPStatusError as exc:
+        msg = f"MIT PE webhook failed with status {exc.response.status_code}: {exc}"
+        context.log.exception(msg)
+        raise RuntimeError(msg) from exc
+
+    context.add_output_metadata(
+        {
+            "delivered_count": len(resources),
+            "webhook_status": "success",
+            "response": MetadataValue.json(response),
+        }
+    )
+    return {"delivered_count": len(resources), "webhook_status": "success"}