diff --git a/env/backend.local.example.env b/env/backend.local.example.env index bdc13bc00d..4ea0e69071 100644 --- a/env/backend.local.example.env +++ b/env/backend.local.example.env @@ -7,6 +7,13 @@ # AWS_STORAGE_BUCKET_NAME= # YOUTUBE_DEVELOPER_KEY= # +# WAREHOUSE_BACKEND=trino +# TRINO_HOST= +# TRINO_PORT=443 +# TRINO_USER= +# TRINO_PASSWORD= +# TRINO_CATALOG=ol_warehouse_production +# # SOCIAL_AUTH_OL_OIDC_OIDC_ENDPOINT= # SOCIAL_AUTH_OL_OIDC_KEY= # SOCIAL_AUTH_OL_OIDC_SECRET= diff --git a/learning_resources/etl/catalog_sources.py b/learning_resources/etl/catalog_sources.py new file mode 100644 index 0000000000..edd0506322 --- /dev/null +++ b/learning_resources/etl/catalog_sources.py @@ -0,0 +1,379 @@ +"""Transform functions for Cohort 1 warehouse-pull catalog sources. + +Each ``transform_*`` function maps one row of an +``integrations__learn__*`` warehouse view (queried via +``learning_resources.lib.warehouse``; see ol-data-platform's +``src/ol_dbt/models/integrations/learn/`` for the view definitions) into the +course/program dict shape expected by ``learning_resources.etl.loaders``. +These functions operate on plain row dicts and have no dependency on the +query engine (Trino today) behind them. The views deliberately expose a +flattened contract (delimited strings instead of nested JSON) so detail +that the platform can't or doesn't compute — per-run pricing, certification +eligibility, program requirement trees — is intentionally left untouched +(via loaders.load_run's `None`-means-"not provided" sentinel, not `[]`) +rather than reconstructed here, so a sync doesn't wipe values the +API-based ETL already set. +""" + +from datetime import UTC + +from dateutil.parser import parse + +from learning_resources.constants import ( + LearningResourceType, + OfferedBy, + PlatformType, +) +from learning_resources.etl.constants import XPRO_PLATFORM_TRANSFORM, ETLSource +from learning_resources.etl.micromasters import ( + READABLE_ID_PREFIX as MICROMASTERS_READABLE_ID_PREFIX, +) +from learning_resources.etl.utils import ( + generate_course_numbers_json, + get_department_id_by_name, + transform_topics, +) +from main.utils import clean_data + + +def _split(value: str | None, sep: str = ",") -> list[str]: + """Split a delimited string column into stripped, non-empty parts. + + Defaults to a plain comma (matching the split convention used elsewhere + in learning_resources.etl, e.g. ocw.py/podcast.py/sloan.py) rather than + the ol-data-platform views' actual ``", "`` array_join separator — + per-part ``.strip()`` makes the two equivalent for well-formed input, + and the plain comma is also correct if a view ever emits "a,b" without + the space. + """ + if not value: + return [] + return [part.strip() for part in value.split(sep) if part.strip()] + + +def _parse_bool(value) -> bool: + # Trino returns native bools; StarRocks (MySQL wire protocol, the + # planned next backend per learning_resources.lib.warehouse) returns + # 1/0 for BOOLEAN columns, which `str(1).lower() == "true"` would + # silently treat as False for every row. + if isinstance(value, bool | int): + return bool(value) + return str(value).strip().lower() == "true" + + +def _parse_datetime(value): + """Parse a warehouse datetime string (matches mitxonline.py/xpro.py).""" + if not value: + return None + try: + return parse(value).replace(tzinfo=UTC) + except (ValueError, TypeError, OverflowError): + return None + + +def _image(url: str | None) -> dict | None: + return {"url": url} if url else None + + +def _topics(value: str | None, offeror_code: str) -> list[dict]: + return transform_topics([{"name": name} for name in _split(value)], offeror_code) + + +def _instructors(value: str | None) -> list[dict]: + return [{"full_name": name} for name in _split(value)] + + +def _parse_runs(value: str | None, title: str, instructors: list[dict]) -> list[dict]: + """Parse ``run_id|start_on|end_on|is_live`` strings joined by ``;``.""" + runs = [] + for chunk in _split(value, sep=";"): + parts = chunk.split("|") + if len(parts) != 4: # noqa: PLR2004 + continue + run_id, start_on, end_on, is_live = parts + if not run_id: + continue + runs.append( + { + "run_id": run_id, + "title": title, + "published": _parse_bool(is_live), + "start_date": _parse_datetime(start_on), + "end_date": _parse_datetime(end_on), + "instructors": instructors, + # The views expose no pricing data. `None` (loaders.load_run's + # sentinel for "not provided, leave alone") rather than `[]` + # so a warehouse sync doesn't wipe prices the API-based ETL + # already populated for this run. + "prices": None, + } + ) + return runs + + +def _course_stub(readable_id: str, platform: str) -> dict: + """Minimal course dict for program->course linking (config.fetch_only=True).""" + return {"readable_id": readable_id, "platform": platform} + + +def _program_run(run_id: str, title: str, row: dict) -> dict: + return { + "run_id": run_id, + "title": title, + "published": _parse_bool(row.get("published")), + "url": row.get("url"), + # No instructor/price data in the views for program runs either; + # see the comment on `_parse_runs`'s "prices" key above. + "instructors": None, + "prices": None, + } + + +# --------------------------------------------------------------------------- +# MITx Online +# --------------------------------------------------------------------------- + + +def transform_mitxonline_course(row: dict) -> dict: + """Transform an integrations__learn__mitxonline_courses row.""" + title = row["title"] + instructors = _instructors(row.get("instructors")) + return { + "readable_id": row["readable_id"], + "platform": PlatformType.mitxonline.name, + "etl_source": ETLSource.mitxonline.name, + "resource_type": LearningResourceType.course.name, + "title": title, + "offered_by": {"code": OfferedBy.mitx.name}, + "topics": _topics(row.get("topics"), OfferedBy.mitx.name), + "runs": _parse_runs(row.get("runs"), title, instructors), + "course": { + "course_numbers": generate_course_numbers_json( + row["readable_id"], is_ocw=False + ), + }, + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +def transform_mitxonline_program(row: dict) -> dict: + """Transform an integrations__learn__mitxonline_programs row.""" + title = row["title"] + return { + "readable_id": row["readable_id"], + "platform": PlatformType.mitxonline.name, + "etl_source": ETLSource.mitxonline.name, + "resource_type": LearningResourceType.program.name, + "title": title, + "offered_by": {"code": OfferedBy.mitx.name}, + "topics": _topics(row.get("topics"), OfferedBy.mitx.name), + "courses": [ + _course_stub(course_id, PlatformType.mitxonline.name) + for course_id in _split(row.get("courses")) + ], + "runs": [_program_run(row["readable_id"], title, row)], + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +# --------------------------------------------------------------------------- +# xPRO +# --------------------------------------------------------------------------- + + +def _xpro_platform(row: dict) -> str: + return XPRO_PLATFORM_TRANSFORM.get(row.get("platform"), PlatformType.xpro.name) + + +def transform_xpro_course(row: dict) -> dict: + """Transform an integrations__learn__xpro_courses row.""" + title = row["title"] + instructors = _instructors(row.get("instructors")) + platform = _xpro_platform(row) + return { + "readable_id": row["readable_id"], + "platform": platform, + "etl_source": ETLSource.xpro.name, + "resource_type": LearningResourceType.course.name, + "title": title, + "offered_by": {"code": OfferedBy.xpro.name}, + "topics": _topics(row.get("topics"), OfferedBy.xpro.name), + "runs": _parse_runs(row.get("runs"), title, instructors), + "course": { + "course_numbers": generate_course_numbers_json( + row["readable_id"], is_ocw=False + ), + }, + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +def transform_xpro_program(row: dict) -> dict: + """Transform an integrations__learn__xpro_programs row.""" + title = row["title"] + platform = _xpro_platform(row) + return { + "readable_id": row["readable_id"], + "platform": platform, + "etl_source": ETLSource.xpro.name, + "resource_type": LearningResourceType.program.name, + "title": title, + "offered_by": {"code": OfferedBy.xpro.name}, + "topics": _topics(row.get("topics"), OfferedBy.xpro.name), + "courses": [ + _course_stub(course_id, platform) + for course_id in _split(row.get("courses")) + ], + "runs": [_program_run(row["readable_id"], title, row)], + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +# --------------------------------------------------------------------------- +# MIT edX (edx.org) +# --------------------------------------------------------------------------- + + +def transform_mit_edx_course(row: dict) -> dict: + """Transform an integrations__learn__mit_edx_courses row.""" + title = row["title"] + instructors = _instructors(row.get("instructors")) + return { + "readable_id": row["readable_id"], + "platform": PlatformType.edx.name, + "etl_source": ETLSource.mit_edx.name, + "resource_type": LearningResourceType.course.name, + "title": title, + "offered_by": {"code": OfferedBy.mitx.name}, + "topics": _topics(row.get("topics"), OfferedBy.mitx.name), + "runs": _parse_runs(row.get("runs"), title, instructors), + "course": { + "course_numbers": generate_course_numbers_json( + row["readable_id"], is_ocw=False + ), + }, + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +# --------------------------------------------------------------------------- +# OCW +# --------------------------------------------------------------------------- + + +def transform_ocw_course(row: dict) -> dict: + """Transform an integrations__learn__ocw_courses row. + + OCW has no run-level data in the view (one row per course); a single + synthetic run is built from the course itself, matching the run OCW's + API-based ETL (learning_resources.etl.ocw.transform_course) constructs. + """ + title = row["title"] + readable_id = row["readable_id"] + published = _parse_bool(row.get("published")) + extra_course_numbers = _split(row.get("extra_course_numbers"), sep=",") + departments = [ + department_id + for department_id in ( + get_department_id_by_name(name) for name in _split(row.get("departments")) + ) + if department_id + ] + return { + "readable_id": readable_id, + "platform": PlatformType.ocw.name, + "etl_source": ETLSource.ocw.name, + "resource_type": LearningResourceType.course.name, + "title": title, + "offered_by": {"code": OfferedBy.ocw.name}, + "topics": _topics(row.get("topics"), OfferedBy.ocw.name), + "instructors": _instructors(row.get("instructors")), + "departments": departments, + "course": { + "course_numbers": generate_course_numbers_json( + row.get("course_number") or readable_id, + extra_nums=extra_course_numbers, + is_ocw=True, + ), + }, + "runs": [ + { + "run_id": readable_id, + "title": title, + "published": published, + "url": row.get("url"), + "semester": row.get("term"), + # LearningResourceRun.year is an IntegerField; an empty + # string coerces fine as "" -> falsy -> None below, but + # would otherwise raise on save. + "year": int(row["year"]) if row.get("year") else None, + "instructors": _instructors(row.get("instructors")), + } + ], + "published": published, + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } + + +# --------------------------------------------------------------------------- +# MicroMasters +# --------------------------------------------------------------------------- + + +def transform_micromasters_program(row: dict) -> dict: + """Transform an integrations__learn__micromasters_programs row. + + MicroMasters courses are hosted on edX; child course readable_ids are + the edX course keys, so program->course linking matches on + ``PlatformType.edx``, same as the API-based MicroMasters ETL. + """ + title = row["title"] + # The view emits the bare `program_id` (e.g. "3"); the API-based ETL + # (learning_resources.etl.micromasters) prefixes it to + # "micromasters-program-3", which is what's already in prod. Passing + # the bare id through would create a duplicate program and the prune + # step would then unpublish the real one. + readable_id = f"{MICROMASTERS_READABLE_ID_PREFIX}{row['readable_id']}" + return { + "readable_id": readable_id, + "platform": PlatformType.edx.name, + "etl_source": ETLSource.micromasters.name, + "resource_type": LearningResourceType.program.name, + "title": title, + "offered_by": {"code": OfferedBy.mitx.name}, + # integrations__learn__micromasters_programs exposes no topics + # column (unlike the other Cohort 1 views) — omitting this key + # entirely would still default to [] in loaders.load_program's + # `program_data.pop("topics", [])`, which load_topics treats as + # "clear all topics", wiping out whatever's curated today on every + # sync. `None` is loaders.py's existing sentinel for "not provided, + # leave alone" (see the analogous `departments_data = ... pop(..., + # None)` a few lines below load_program's topics handling). + "topics": None, + "courses": [ + _course_stub(course_id, PlatformType.edx.name) + for course_id in _split(row.get("courses")) + ], + "runs": [_program_run(readable_id, title, row)], + "published": _parse_bool(row.get("published")), + "image": _image(row.get("image_url")), + "url": row.get("url"), + "description": clean_data(row.get("description")), + } diff --git a/learning_resources/etl/catalog_sources_test.py b/learning_resources/etl/catalog_sources_test.py new file mode 100644 index 0000000000..9d2a69ea39 --- /dev/null +++ b/learning_resources/etl/catalog_sources_test.py @@ -0,0 +1,336 @@ +"""Tests for learning_resources.etl.catalog_sources transform functions.""" + +import pytest + +from learning_resources.constants import LearningResourceType, OfferedBy, PlatformType +from learning_resources.etl import catalog_sources +from learning_resources.etl.constants import ETLSource + +pytestmark = pytest.mark.django_db + + +def test_split_defaults_and_strips(): + """_split returns [] for empty input and strips/filters delimited parts.""" + assert catalog_sources._split(None) == [] # noqa: SLF001 + assert catalog_sources._split("") == [] # noqa: SLF001 + assert catalog_sources._split("a, b, c ") == ["a", "b", "c"] # noqa: SLF001 + assert catalog_sources._split("a;b", sep=";") == ["a", "b"] # noqa: SLF001 + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (True, True), + (False, False), + ("true", True), + ("True", True), + ("false", False), + ("", False), + (None, False), + (1, True), + (0, False), + ], +) +def test_parse_bool(value, expected): + """_parse_bool handles native bools, ints (StarRocks), and 'true'/'false' text.""" + assert catalog_sources._parse_bool(value) is expected # noqa: SLF001 + + +def test_parse_datetime_valid_and_invalid(): + """_parse_datetime parses ISO strings and returns None for bad input.""" + assert catalog_sources._parse_datetime("2026-01-01T00:00:00").year == 2026 # noqa: SLF001 + assert catalog_sources._parse_datetime("") is None # noqa: SLF001 + assert catalog_sources._parse_datetime(None) is None # noqa: SLF001 + assert catalog_sources._parse_datetime("not-a-date") is None # noqa: SLF001 + + +def test_parse_runs_splits_and_parses_pipe_delimited_string(): + """_parse_runs splits ';'-joined 'id|start|end|is_live' run strings.""" + runs = catalog_sources._parse_runs( # noqa: SLF001 + "run-1|2026-01-01|2026-06-01|true;run-2|2026-07-01||false", + "Course Title", + instructors=[{"full_name": "Jane Doe"}], + ) + assert len(runs) == 2 + assert runs[0]["run_id"] == "run-1" + assert runs[0]["title"] == "Course Title" + assert runs[0]["published"] is True + assert runs[0]["start_date"].year == 2026 + assert runs[0]["end_date"].year == 2026 + assert runs[0]["instructors"] == [{"full_name": "Jane Doe"}] + assert runs[0]["prices"] is None + assert runs[1]["run_id"] == "run-2" + assert runs[1]["published"] is False + assert runs[1]["end_date"] is None + + +def test_parse_runs_prices_is_none_not_empty_list(): + """prices=None (not []) so loaders.load_run's None-sentinel leaves + existing run prices alone — the views have no pricing data, and a [] + here would wipe out prices the API-based ETL already set on this run. + """ + runs = catalog_sources._parse_runs( # noqa: SLF001 + "run-1|2026-01-01|2026-06-01|true", "T", instructors=[] + ) + assert runs[0]["prices"] is None + + +def test_parse_runs_skips_malformed_chunks(): + """_parse_runs drops chunks that don't have exactly 4 pipe-delimited fields.""" + runs = catalog_sources._parse_runs( # noqa: SLF001 + "missing-fields|only-two", "T", instructors=[] + ) + assert runs == [] + + +def test_parse_runs_empty_value(): + """_parse_runs returns [] for a missing/empty runs column.""" + assert catalog_sources._parse_runs(None, "T", instructors=[]) == [] # noqa: SLF001 + + +MITXONLINE_COURSE_ROW = { + "readable_id": "course-v1:MITxT+1.1x", + "title": "Intro to Testing", + "last_modified": "2026-01-01T00:00:00", + "etl_source": "mitxonline", + "description": "A course about testing", + "url": "/courses/intro-to-testing", + "image_url": "https://example.com/image.jpg", + "published": True, + "platform": "mitxonline", + "page_slug": "intro-to-testing", + "topics": "Testing, Quality", + "instructors": "Jane Doe, John Smith", + "certification_type": "professional", + "price": "49.00", + "length": "6 weeks", + "effort": "5-10 hours/week", + "runs": "course-v1:MITxT+1.1x+R1|2026-01-01|2026-06-01|true", +} + + +def test_transform_mitxonline_course(): + """transform_mitxonline_course maps a view row to the loader course shape.""" + result = catalog_sources.transform_mitxonline_course(MITXONLINE_COURSE_ROW) + + assert result["readable_id"] == MITXONLINE_COURSE_ROW["readable_id"] + assert result["platform"] == PlatformType.mitxonline.name + assert result["etl_source"] == ETLSource.mitxonline.name + assert result["resource_type"] == LearningResourceType.course.name + assert result["title"] == "Intro to Testing" + assert result["offered_by"] == {"code": OfferedBy.mitx.name} + assert result["published"] is True + assert result["image"] == {"url": "https://example.com/image.jpg"} + assert len(result["runs"]) == 1 + assert result["runs"][0]["run_id"] == "course-v1:MITxT+1.1x+R1" + assert result["course"]["course_numbers"] + + +MITXONLINE_PROGRAM_ROW = { + "readable_id": "program-v1:MITxT+PP", + "title": "Professional Program", + "etl_source": "mitxonline", + "description": "A program", + "url": "/programs/professional-program", + "image_url": "https://example.com/program.jpg", + "published": True, + "topics": "Testing", + "courses": "course-v1:MITxT+1.1x, course-v1:MITxT+1.2x", +} + + +def test_transform_mitxonline_program(): + """transform_mitxonline_program builds fetch_only course stubs and a single run.""" + result = catalog_sources.transform_mitxonline_program(MITXONLINE_PROGRAM_ROW) + + assert result["resource_type"] == LearningResourceType.program.name + assert result["courses"] == [ + { + "readable_id": "course-v1:MITxT+1.1x", + "platform": PlatformType.mitxonline.name, + }, + { + "readable_id": "course-v1:MITxT+1.2x", + "platform": PlatformType.mitxonline.name, + }, + ] + assert len(result["runs"]) == 1 + assert result["runs"][0]["run_id"] == MITXONLINE_PROGRAM_ROW["readable_id"] + # No instructor/price data for program runs in the views either; None + # (not []) so loaders.load_run leaves any existing values alone. + assert result["runs"][0]["instructors"] is None + assert result["runs"][0]["prices"] is None + + +@pytest.mark.parametrize( + ("platform_name", "expected"), + [ + ("xPRO", PlatformType.xpro.name), + ("Emeritus", PlatformType.emeritus.name), + ("Unknown Platform", PlatformType.xpro.name), + (None, PlatformType.xpro.name), + ], +) +def test_transform_xpro_course_platform_mapping(platform_name, expected): + """transform_xpro_course maps the view's display name via XPRO_PLATFORM_TRANSFORM.""" + row = { + "readable_id": "course-v1:xPRO+1x", + "title": "xPRO Course", + "platform": platform_name, + "published": True, + "topics": "", + "instructors": "", + "runs": "", + } + result = catalog_sources.transform_xpro_course(row) + assert result["platform"] == expected + assert result["etl_source"] == ETLSource.xpro.name + + +def test_transform_xpro_program_courses_use_resolved_platform(): + """transform_xpro_program's course stubs use the program's resolved platform.""" + row = { + "readable_id": "program-v1:xPRO+PP", + "title": "xPRO Program", + "platform": "Emeritus", + "published": True, + "topics": "", + "courses": "course-v1:xPRO+1x", + } + result = catalog_sources.transform_xpro_program(row) + assert result["courses"] == [ + {"readable_id": "course-v1:xPRO+1x", "platform": PlatformType.emeritus.name} + ] + + +def test_transform_mit_edx_course_uses_platform_edx_not_view_value(): + """transform_mit_edx_course uses PlatformType.edx, not the view's raw 'edxorg' value.""" + row = { + "readable_id": "MITx/6.00.1x", + "title": "MIT edX Course", + "platform": "edxorg", # not a valid PlatformType member + "published": True, + "topics": "CS, Programming", + "instructors": "Ada Lovelace", + "runs": "MITx/6.00.1x/run|2026-01-01|2026-06-01|true", + } + result = catalog_sources.transform_mit_edx_course(row) + assert result["platform"] == PlatformType.edx.name + assert result["etl_source"] == ETLSource.mit_edx.name + assert result["runs"][0]["instructors"] == [{"full_name": "Ada Lovelace"}] + + +def test_transform_ocw_course_single_synthetic_run_and_departments(): + """transform_ocw_course builds one synthetic run and resolves department names to ids.""" + row = { + "readable_id": "6-00-1x-fall-2025", + "title": "OCW Course", + "description": "desc", + "url": "https://ocw.mit.edu/6-00-1x", + "image_url": "https://ocw.mit.edu/image.jpg", + "published": True, + "level": "Undergraduate", + "term": "Fall", + "year": "2025", + "course_number": "6.00.1x", + "extra_course_numbers": "6.01,6.02", + "topics": "Programming, Algorithms", + "instructors": "Grace Hopper", + "departments": "Mathematics, Physics", + } + result = catalog_sources.transform_ocw_course(row) + + assert result["platform"] == PlatformType.ocw.name + assert result["etl_source"] == ETLSource.ocw.name + assert result["departments"] == ["18", "8"] + assert len(result["runs"]) == 1 + run = result["runs"][0] + assert run["run_id"] == row["readable_id"] + assert run["semester"] == "Fall" + assert run["year"] == 2025 + assert run["instructors"] == [{"full_name": "Grace Hopper"}] + assert result["course"]["course_numbers"] + + +def test_transform_ocw_course_empty_year_coerces_to_none(): + """An empty-string year (LearningResourceRun.year is an IntegerField) + coerces to None instead of raising on save. + """ + row = { + "readable_id": "6-00-1x-fall-2025", + "title": "OCW Course", + "published": True, + "term": "Fall", + "year": "", + } + result = catalog_sources.transform_ocw_course(row) + assert result["runs"][0]["year"] is None + + +def test_transform_ocw_course_unknown_department_is_dropped(): + """transform_ocw_course silently drops department names with no known id.""" + row = { + "readable_id": "6-00-1x-fall-2025", + "title": "OCW Course", + "published": True, + "departments": "Not A Real Department", + } + result = catalog_sources.transform_ocw_course(row) + assert result["departments"] == [] + + +def test_transform_micromasters_program_uses_edx_platform_for_courses(): + """transform_micromasters_program's child courses resolve on PlatformType.edx.""" + row = { + "readable_id": "1", + "title": "MicroMasters Program", + "published": True, + "url": "https://micromasters.mit.edu/1", + "courses": "MITx/1.1x, MITx/1.2x", + } + result = catalog_sources.transform_micromasters_program(row) + + assert result["platform"] == PlatformType.edx.name + assert result["etl_source"] == ETLSource.micromasters.name + assert result["courses"] == [ + {"readable_id": "MITx/1.1x", "platform": PlatformType.edx.name}, + {"readable_id": "MITx/1.2x", "platform": PlatformType.edx.name}, + ] + + +def test_transform_micromasters_program_prefixes_readable_id(): + """readable_id/run_id get the same "micromasters-program-" prefix the + API-based ETL (learning_resources.etl.micromasters) uses, since the + view emits the bare program_id — passing it through unprefixed would + create a duplicate program and the prune step would unpublish the real + one (prod already has "micromasters-program-1"). + """ + row = { + "readable_id": "1", + "title": "MicroMasters Program", + "published": True, + "url": "https://micromasters.mit.edu/1", + "courses": "", + } + result = catalog_sources.transform_micromasters_program(row) + + assert result["readable_id"] == "micromasters-program-1" + assert result["runs"][0]["run_id"] == "micromasters-program-1" + + +def test_transform_micromasters_program_topics_is_none_not_empty_list(): + """topics=None (not []) so loaders.load_program's load_topics skips and + leaves existing topics alone — the view has no topics column, and a [] + here would make load_topics call resource.topics.set([]), silently + wiping out whatever's curated today on every sync. + """ + row = { + "readable_id": "1", + "title": "MicroMasters Program", + "published": True, + "url": "https://micromasters.mit.edu/1", + "courses": "", + } + result = catalog_sources.transform_micromasters_program(row) + + assert result["topics"] is None diff --git a/learning_resources/etl/constants.py b/learning_resources/etl/constants.py index 7f6351e55e..133f73112b 100644 --- a/learning_resources/etl/constants.py +++ b/learning_resources/etl/constants.py @@ -9,7 +9,7 @@ from django.conf import settings from named_enum import ExtendedEnum -from learning_resources.constants import LearningResourceDelivery +from learning_resources.constants import LearningResourceDelivery, PlatformType from learning_resources.models import LearningResourcePrice # A custom UA so that operators of OpenEdx will know who is pinging their service @@ -21,6 +21,18 @@ MIT_OWNER_KEYS = ["MITx", "MITx_PRO"] +# Shared by xpro.py (API-based ETL) and catalog_sources.py (warehouse-pull +# ETL) — both need to map xPRO's CMS platform display name to a PlatformType. +# This needs to be kept up to date with valid xpro platforms. +XPRO_PLATFORM_TRANSFORM = { + "Emeritus": PlatformType.emeritus.name, + "Global Alumni": PlatformType.globalalumni.name, + "Simplilearn": PlatformType.simplilearn.name, + "Susskind": PlatformType.susskind.name, + "WHU": PlatformType.whu.name, + "xPRO": PlatformType.xpro.name, +} + TIME_INTERVAL_MAPPING = { "half-days": ["days"], "half-day": ["day"], diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index fedaf7a3d6..689e46eb79 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -193,9 +193,17 @@ def load_run_dependent_values( def load_instructors( - run: LearningResourceRun, instructors_data: list[dict] + run: LearningResourceRun, instructors_data: list[dict] | None ) -> list[LearningResourceInstructor]: - """Load the instructors for a resource run into the database""" + """Load the instructors for a resource run into the database. + + `None` (as opposed to `[]`) means the source didn't provide instructor + data at all; leave whatever's already on the run alone rather than + clearing it — same convention as load_topics's `topics_data`. + """ + if instructors_data is None: + return list(run.instructors.all()) + instructors = [] valid_attributes = ["first_name", "last_name"] relations = [] @@ -227,9 +235,17 @@ def load_instructors( def load_prices( - run: LearningResourceRun, prices_data: list[dict] + run: LearningResourceRun, prices_data: list[dict] | None ) -> list[LearningResourcePrice]: - """Load the prices for a resource run into the database""" + """Load the prices for a resource run into the database. + + `None` (as opposed to `[]`) means the source didn't provide price data + at all; leave whatever's already on the run alone rather than clearing + it — same convention as load_topics's `topics_data`. + """ + if prices_data is None: + return list(run.resource_prices.all()) + prices = [] for price in prices_data: lr_price, _ = LearningResourcePrice.objects.get_or_create( @@ -305,6 +321,29 @@ def load_content_tags( ) +def _resolve_run_prices( + run_data: dict, + resource_prices: list[dict] | None, + status: str | None, + learning_resource: LearningResource, +) -> list[dict] | None: + """Normalize run_data's "prices" summary field for one run. + + Returns the resource_prices list to pass to load_prices, or None if no + price data was provided by this source (leave existing values alone). + """ + if resource_prices is None: + return None + + run_data["prices"] = sorted({price["amount"] for price in resource_prices}) + if status == RunStatus.archived.value or learning_resource.certification is False: + # Archived runs or runs of resources w/out certificates should not + # have prices + run_data["prices"] = [] + return [] + return resource_prices + + def load_run( learning_resource: LearningResource, run_data: dict ) -> LearningResourceRun: @@ -322,15 +361,16 @@ def load_run( image_data = run_data.pop("image", None) status = run_data.pop("status", None) + # `None` (as opposed to an omitted key, which defaults to `[]`) is the + # sentinel for "not provided by this source, leave existing value + # alone" — same convention load_topics uses for `topics_data`. Sources + # that don't have instructor/price data (e.g. the warehouse-pull + # transforms, which lack pricing entirely) pass `None` explicitly + # rather than `[]`, so a sync doesn't wipe data another pipeline wrote. instructors_data = run_data.pop("instructors", []) - - resource_prices = run_data.get("prices", []) - run_data["prices"] = sorted({price["amount"] for price in resource_prices}) - - if status == RunStatus.archived.value or learning_resource.certification is False: - # Archived runs or runs of resources w/out certificates should not have prices - run_data["prices"] = [] - resource_prices = [] + resource_prices = _resolve_run_prices( + run_data, run_data.pop("prices", []), status, learning_resource + ) if learning_resource.test_mode: run_data["published"] = True diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index c0a195dd43..7e6dac2443 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -50,6 +50,7 @@ load_podcast, load_podcast_episode, load_podcasts, + load_prices, load_problem_file, load_problem_files, load_program, @@ -1026,6 +1027,40 @@ def test_load_run(mocker, run_exists, status, certification): mock_import_task.delay.assert_not_called() +@pytest.mark.django_db(transaction=True) +def test_load_run_none_instructors_and_prices_leave_existing_values_alone(mocker): + """`None` (not an omitted key, which still defaults to `[]`) for + run_data's "instructors"/"prices" is the sentinel for "not provided by + this source, leave alone" — same convention load_topics already uses + for `topics_data`. A source that doesn't have instructor/price data + (e.g. the warehouse-pull transforms) must not wipe out values another + pipeline already populated. + """ + mocker.patch("learning_resources.tasks.import_content_files") + course = LearningResourceFactory.create( + is_course=True, runs=[], certification=True, etl_source=ETLSource.xpro.value + ) + run = LearningResourceRunFactory.create(learning_resource=course, prices=[]) + instructor = LearningResourceInstructorFactory.create(full_name="Jane Doe") + load_instructors(run, [{"full_name": instructor.full_name}]) + load_prices(run, [{"amount": Decimal("49.00"), "currency": CURRENCY_USD}]) + run.prices = [Decimal("49.00")] + run.save() + + run_data = { + "run_id": run.run_id, + "title": run.title, + "instructors": None, + "prices": None, + } + result = load_run(course, run_data) + + assert result.id == run.id + assert [i.full_name for i in result.instructors.all()] == [instructor.full_name] + assert [p.amount for p in result.resource_prices.all()] == [Decimal("49.00")] + assert result.prices == [Decimal("49.00")] + + @pytest.mark.parametrize( "etl_source", [ETLSource.mit_edx.value, ETLSource.mitxonline.value, ETLSource.xpro.value], diff --git a/learning_resources/etl/xpro.py b/learning_resources/etl/xpro.py index ba78b8f293..2930a195ab 100644 --- a/learning_resources/etl/xpro.py +++ b/learning_resources/etl/xpro.py @@ -15,9 +15,8 @@ LearningResourceType, OfferedBy, Pace, - PlatformType, ) -from learning_resources.etl.constants import ETLSource +from learning_resources.etl.constants import XPRO_PLATFORM_TRANSFORM, ETLSource from learning_resources.etl.utils import ( generate_course_numbers_json, parse_string_to_int, @@ -31,16 +30,6 @@ OFFERED_BY = {"code": OfferedBy.xpro.name} -# This needs to be kept up to date with valid xpro platforms -XPRO_PLATFORM_TRANSFORM = { - "Emeritus": PlatformType.emeritus.name, - "Global Alumni": PlatformType.globalalumni.name, - "Simplilearn": PlatformType.simplilearn.name, - "Susskind": PlatformType.susskind.name, - "WHU": PlatformType.whu.name, - "xPRO": PlatformType.xpro.name, -} - def _parse_datetime(value): """ diff --git a/learning_resources/lib/__init__.py b/learning_resources/lib/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/learning_resources/lib/warehouse.py b/learning_resources/lib/warehouse.py new file mode 100644 index 0000000000..3d31d369bd --- /dev/null +++ b/learning_resources/lib/warehouse.py @@ -0,0 +1,291 @@ +"""SQL warehouse pull infrastructure for data platform consumption. + +Query engine is not hard-wired into the ETL tasks or row-iteration logic: +the concrete backend is selected at connect time via ``settings.WAREHOUSE_BACKEND``. +Trino is the only implemented backend today; the OL Data Platform may migrate +this endpoint to StarRocks later, and since both expose a standard Python +DB-API 2.0 cursor (``execute`` / ``fetchmany`` / ``description``), adding that +backend should only mean filling in ``_connect_starrocks`` below — ``iter_rows`` +and ``BaseWarehouseETLTask`` need no changes. +""" + +import logging +import re +import time +from datetime import UTC, datetime + +import sentry_sdk +from celery import Task +from django.conf import settings +from django.core.cache import caches +from django.core.exceptions import ImproperlyConfigured + +log = logging.getLogger(__name__) + +# Only alphanumeric, underscores, and dots (schema.table notation). `\Z` +# (not `$`) is deliberate: `$` matches just before a trailing newline, which +# would let a `view_name` ending in "\n" slip through this check. +_SAFE_IDENTIFIER = re.compile(r"^[a-zA-Z0-9_.]+\Z") + +# Cache alias for sync watermarks: must outlive Redis flushes (an evicted +# watermark just makes the next incremental run fetch a wider-than-necessary +# window, which is safe — see BaseWarehouseETLTask.fetch_and_upsert's +# ``since`` contract), but should not require a dedicated model + migration +# for what is ETL bookkeeping, not domain data. +_WATERMARK_CACHE = "durable" +_WATERMARK_SQL_FORMAT = "%Y-%m-%d %H:%M:%S.%f" + + +def _connect_trino(): + """Open a Trino DB-API connection using Django settings. + + Raises: + ImproperlyConfigured: If TRINO_HOST or TRINO_USER is unset — fails + fast with a clear message instead of BasicAuthentication + silently pairing None/None credentials and connect() attempting + host=None/user=None (Trino requires a non-empty user even for + unauthenticated connections). + """ + from trino.auth import BasicAuthentication + from trino.dbapi import connect + + if not settings.TRINO_HOST or not settings.TRINO_USER: + msg = ( + "WAREHOUSE_BACKEND is 'trino' but TRINO_HOST/TRINO_USER are not " + "set. Set TRINO_HOST/TRINO_USER/TRINO_PASSWORD/TRINO_CATALOG, " + "or switch WAREHOUSE_BACKEND to a configured backend." + ) + raise ImproperlyConfigured(msg) + + auth = ( + BasicAuthentication(settings.TRINO_USER, settings.TRINO_PASSWORD) + if settings.TRINO_USER and settings.TRINO_PASSWORD + else None + ) + return connect( + host=settings.TRINO_HOST, + port=settings.TRINO_PORT, + user=settings.TRINO_USER, + auth=auth, + catalog=settings.TRINO_CATALOG, + ) + + +def _connect_starrocks(): + """Open a StarRocks DB-API connection using Django settings. + + Not yet implemented — no StarRocks endpoint is provisioned for MIT + Learn. StarRocks speaks the MySQL wire protocol and the same DB-API 2.0 + cursor surface Trino does, so this should be a drop-in connect() call + (e.g. via ``mysql-connector-python`` or ``starrocks``) once one exists. + """ + msg = "The starrocks warehouse backend is not yet implemented" + raise NotImplementedError(msg) + + +_CONNECTORS = { + "trino": _connect_trino, + "starrocks": _connect_starrocks, +} + + +def connect_to_warehouse(): + """Open and return a DB-API connection for the configured warehouse backend. + + Returns: + A DB-API 2.0 connection (``trino.dbapi.Connection`` today). + + Raises: + ValueError: If ``settings.WAREHOUSE_BACKEND`` names an unknown backend. + Exception: propagated if the connection attempt fails. + """ + backend = settings.WAREHOUSE_BACKEND + try: + connector = _CONNECTORS[backend] + except KeyError: + msg = f"Unknown WAREHOUSE_BACKEND: {backend!r}" + raise ValueError(msg) from None + + try: + conn = connector() + except Exception: + log.exception("Failed to connect to warehouse backend %s", backend) + raise + log.info("Connected to warehouse backend %s", backend) + return conn + + +def iter_rows(conn, view_name, *, since=None, batch_size=1000): + """Iterate over rows in a warehouse view as column-keyed dicts. + + Backend-agnostic: relies only on the DB-API 2.0 cursor surface + (``execute`` / ``description`` / ``fetchmany``), which Trino and + StarRocks both implement. + + Args: + conn: An open DB-API connection. + view_name (str): Schema-qualified view name, e.g. + ``"integrations.integrations__learn__ocw_courses"`` — not + catalog-qualified; the connection's configured catalog (e.g. + ``settings.TRINO_CATALOG``) supplies that. A fully-qualified + ``catalog.schema.table`` name also works if callers want to + override the connection's default catalog for one query. Must + contain only alphanumeric characters, underscores, and dots. + since (datetime | None): If given, only rows whose ``last_modified`` + column is greater than this timestamp are returned — every + ``integrations__learn__*`` view exposes ``last_modified`` per the + contract in ol-data-platform's docs/learn_marts_contract.md, so + this is safe to assume for any view passed here. ``None`` (the + default) pulls every row. + batch_size (int): Rows fetched per round-trip. + + Yields: + dict: One row, keyed by column name. + + Raises: + ValueError: If *view_name* contains characters that could allow SQL injection. + """ + if not _SAFE_IDENTIFIER.match(view_name): + msg = f"Unsafe view name: {view_name!r}" + raise ValueError(msg) + + query = f"SELECT * FROM {view_name}" # noqa: S608 + if since is not None: + # `since` is produced by our own watermark tracking, never user + # input, so a formatted literal is safe here (no parameterization + # needed) — and `TIMESTAMP '...'` literal syntax is understood by + # Trino, StarRocks/MySQL, and DuckDB alike. + watermark = since.strftime(_WATERMARK_SQL_FORMAT)[:-3] + query += f" WHERE last_modified > TIMESTAMP '{watermark}'" + + cur = conn.cursor() + try: + cur.execute(query) + # Read columns from the first non-empty batch rather than + # immediately after execute(): some DB-API drivers only populate + # `description` once results start arriving, not at execute() + # time (PEP 249 leaves this driver-defined). + columns = None + while True: + batch = cur.fetchmany(batch_size) + if not batch: + break + if columns is None: + columns = [d[0] for d in cur.description] + for row in batch: + yield dict(zip(columns, row)) + finally: + cur.close() + + +class BaseWarehouseETLTask(Task): + """Celery task base class for warehouse-pull ETL jobs. + + Subclasses declare ``view_name`` and implement + ``fetch_and_upsert(conn, *, since)``. Register concrete subclasses via + ``app.register_task(SubclassTask())`` — a class decorated with + ``@app.task(base=BaseWarehouseETLTask)`` does *not* work: Celery's + function-task machinery wraps the decorated object as a ``staticmethod`` + and calls it positionally, which — given a class — tries to instantiate + it as the task body instead of running its bound + ``run()``/``fetch_and_upsert()`` methods. + + Two sync modes, selected by the ``full_refresh`` kwarg passed to + ``run()``/``.delay()``/``.apply_async()``: + + - ``full_refresh=True`` (the default, and what the daily beat schedule + uses today): ``since`` is ``None`` — ``fetch_and_upsert`` pulls every + row and should prune resources no longer present in the source. This + is the self-healing baseline: it's the only mode that ever sees + deletes/unpublishes upstream. + - ``full_refresh=False``: ``since`` is this task's last recorded + watermark (or ``None`` if it has never completed an incremental run, + in which case it behaves like a full pull). ``fetch_and_upsert`` + should pass ``since`` through to ``iter_rows`` and skip pruning — + a partial pull must never be treated as the complete state of the + source. On success the watermark is advanced to "now". + + Example:: + + class SyncOCWCoursesTask(BaseWarehouseETLTask): + name = "learning_resources.tasks.SyncOCWCoursesTask" + view_name = "integrations.integrations__learn__ocw_courses" + + def fetch_and_upsert(self, conn, *, since=None): + for row in iter_rows(conn, self.view_name, since=since): + upsert_ocw_course(row, prune=since is None) + + SyncOCWCoursesTask = app.register_task(SyncOCWCoursesTask()) + """ + + abstract = True + acks_late = True + view_name: str = "" + + def run(self, *args, full_refresh: bool = True, **kwargs): # noqa: ARG002 + """Open a warehouse connection, delegate to ``fetch_and_upsert``, log counts.""" + if not self.view_name: + msg = f"{self.__class__.__name__}.view_name must be set" + raise ValueError(msg) + + since = None if full_refresh else self._get_watermark() + # Captured before the fetch, not after: a row modified while the + # fetch is in flight must still be picked up by the *next* + # incremental run. Stamping the watermark post-fetch would let it + # fall between this window and the next, invisible until the next + # full_refresh heals it. + fetch_started_at = datetime.now(tz=UTC) + + conn = connect_to_warehouse() + start = time.monotonic() + try: + count = self.fetch_and_upsert(conn, since=since) + except Exception: + log.exception("Warehouse ETL task %s failed", self.name) + sentry_sdk.add_breadcrumb( + category="warehouse_etl", + message=f"{self.name} failed", + data={"view_name": self.view_name, "full_refresh": full_refresh}, + level="error", + ) + raise + finally: + conn.close() + + if not full_refresh: + self._set_watermark(fetch_started_at) + + elapsed = time.monotonic() - start + log.info( + "Warehouse ETL task %s finished (%s): %d rows in %.1fs", + self.name, + "full_refresh" if full_refresh else "incremental", + count, + elapsed, + ) + return count + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull rows from the warehouse and upsert into Django models. + + Args: + conn: An open DB-API connection (will be closed by ``run``). + since (datetime | None): Forwarded from ``run()``. ``None`` means + a full refresh — pull everything and prune. Otherwise, pull + only rows changed since this watermark and skip pruning. + + Returns: + int: Number of rows processed. + """ + raise NotImplementedError + + def _watermark_cache_key(self) -> str: + return f"warehouse_etl:last_synced_at:{self.name}" + + def _get_watermark(self): + """Return this task's last recorded incremental watermark, if any.""" + return caches[_WATERMARK_CACHE].get(self._watermark_cache_key()) + + def _set_watermark(self, value: datetime) -> None: + """Persist ``value`` as this task's incremental watermark.""" + caches[_WATERMARK_CACHE].set(self._watermark_cache_key(), value, timeout=None) diff --git a/learning_resources/lib/warehouse_test.py b/learning_resources/lib/warehouse_test.py new file mode 100644 index 0000000000..1d437b35ea --- /dev/null +++ b/learning_resources/lib/warehouse_test.py @@ -0,0 +1,513 @@ +"""Unit tests for learning_resources.lib.warehouse. + +Intentionally avoids loading the full Django application — patches +django.conf.settings directly so these tests run in milliseconds with no +database setup required. +""" + +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock, patch + +import django +import pytest +from django.conf import settings as django_settings +from freezegun import freeze_time + +# Configure Django minimally if not already configured — keeps this test +# module self-contained and fast (no database, no app registry needed). +if not django_settings.configured: + django_settings.configure( + WAREHOUSE_BACKEND="trino", + TRINO_HOST="trino.example.com", + TRINO_PORT=443, + TRINO_USER="testuser", + TRINO_PASSWORD="secret", # noqa: S106 + TRINO_CATALOG="ol_warehouse_production", + ) + django.setup() + +# `caches["durable"]` (the incremental watermark store) is only touched on +# the full_refresh=False path — every test below mocks it directly rather +# than configuring a real CACHES setting, keeping this module's "no +# database, no app registry" guarantee intact. + + +from learning_resources.lib.warehouse import ( + BaseWarehouseETLTask, + connect_to_warehouse, + iter_rows, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _patch_settings(**kwargs): + """Return a patcher that overrides specific Django settings values.""" + return patch.multiple("django.conf.settings", **kwargs) + + +def _warehouse_settings(**overrides): + settings = { + "WAREHOUSE_BACKEND": "trino", + "TRINO_HOST": "trino.example.com", + "TRINO_PORT": 443, + "TRINO_USER": "testuser", + "TRINO_PASSWORD": "secret", + "TRINO_CATALOG": "ol_warehouse_production", + } + settings.update(overrides) + return settings + + +def _make_cursor(columns, rows): + """Build a minimal mock cursor returning all rows in one batch then empty.""" + cursor = MagicMock() + cursor.description = [(col,) for col in columns] + cursor.fetchmany.side_effect = [rows, []] + return cursor + + +# --------------------------------------------------------------------------- +# connect_to_warehouse +# --------------------------------------------------------------------------- + + +@patch("trino.dbapi.connect") +def test_connect_to_warehouse_trino_uses_settings(mock_connect): + """connect_to_warehouse dispatches to Trino and passes Django settings through.""" + mock_conn = MagicMock() + mock_connect.return_value = mock_conn + + with _patch_settings(**_warehouse_settings()): + result = connect_to_warehouse() + + assert result is mock_conn + call_kwargs = mock_connect.call_args.kwargs + assert call_kwargs["host"] == "trino.example.com" + assert call_kwargs["port"] == 443 + assert call_kwargs["user"] == "testuser" + assert call_kwargs["catalog"] == "ol_warehouse_production" + + +@patch("trino.dbapi.connect", side_effect=OSError("unreachable")) +def test_connect_to_warehouse_propagates_exception(mock_connect): + """connect_to_warehouse re-raises connection errors.""" + with ( + _patch_settings(**_warehouse_settings()), + pytest.raises(OSError, match="unreachable"), + ): + connect_to_warehouse() + + +def test_connect_to_warehouse_rejects_unknown_backend(): + """connect_to_warehouse raises ValueError for an unconfigured backend name.""" + with ( + _patch_settings(**_warehouse_settings(WAREHOUSE_BACKEND="snowflake")), + pytest.raises(ValueError, match="Unknown WAREHOUSE_BACKEND"), + ): + connect_to_warehouse() + + +def test_connect_to_warehouse_starrocks_not_yet_implemented(): + """The starrocks backend is a documented placeholder, not yet wired up.""" + with ( + _patch_settings(**_warehouse_settings(WAREHOUSE_BACKEND="starrocks")), + pytest.raises(NotImplementedError, match="starrocks"), + ): + connect_to_warehouse() + + +# --------------------------------------------------------------------------- +# iter_rows +# --------------------------------------------------------------------------- + + +def test_iter_rows_yields_column_keyed_dicts(): + """iter_rows converts raw tuples to column-keyed dicts.""" + conn = MagicMock() + cursor = _make_cursor(["id", "title"], [(1, "Course A"), (2, "Course B")]) + conn.cursor.return_value = cursor + + rows = list(iter_rows(conn, "catalog.schema.my_view")) + + assert rows == [ + {"id": 1, "title": "Course A"}, + {"id": 2, "title": "Course B"}, + ] + + +def test_iter_rows_respects_batch_size(): + """iter_rows passes batch_size to fetchmany.""" + conn = MagicMock() + cursor = _make_cursor(["id"], [(1,), (2,)]) + conn.cursor.return_value = cursor + + list(iter_rows(conn, "catalog.schema.my_view", batch_size=500)) + + cursor.fetchmany.assert_any_call(500) + + +def test_iter_rows_closes_cursor_on_success(): + """iter_rows always closes the cursor on normal exit.""" + conn = MagicMock() + cursor = _make_cursor(["id"], []) + conn.cursor.return_value = cursor + + list(iter_rows(conn, "catalog.schema.view")) + + cursor.close.assert_called_once() + + +def test_iter_rows_closes_cursor_on_error(): + """iter_rows closes the cursor even when execution raises.""" + conn = MagicMock() + cursor = MagicMock() + cursor.execute.side_effect = RuntimeError("query failed") + conn.cursor.return_value = cursor + + with pytest.raises(RuntimeError): + list(iter_rows(conn, "catalog.schema.view")) + + cursor.close.assert_called_once() + + +@pytest.mark.parametrize( + "bad_name", + [ + "'; DROP TABLE courses; --", + "schema.table; DELETE FROM users", + "schema.table WHERE 1=1", + "../etc/passwd", + pytest.param("schema.table\n", id="trailing-newline"), + ], +) +def test_iter_rows_rejects_unsafe_view_names(bad_name): + """iter_rows raises ValueError for names that could allow SQL injection.""" + conn = MagicMock() + with pytest.raises(ValueError, match="Unsafe view name"): + list(iter_rows(conn, bad_name)) + + +def test_iter_rows_without_since_has_no_where_clause(): + """A full-refresh pull (since=None) queries the view unfiltered.""" + conn = MagicMock() + cursor = _make_cursor(["id"], []) + conn.cursor.return_value = cursor + + list(iter_rows(conn, "catalog.schema.my_view")) + + query = cursor.execute.call_args.args[0] + assert query == "SELECT * FROM catalog.schema.my_view" + + +def test_iter_rows_with_since_filters_on_last_modified(): + """An incremental pull (since=) adds a last_modified predicate.""" + conn = MagicMock() + cursor = _make_cursor(["id"], []) + conn.cursor.return_value = cursor + since = datetime(2026, 6, 15, 12, 30, 0, tzinfo=UTC) + + list(iter_rows(conn, "catalog.schema.my_view", since=since)) + + query = cursor.execute.call_args.args[0] + assert query == ( + "SELECT * FROM catalog.schema.my_view " + "WHERE last_modified > TIMESTAMP '2026-06-15 12:30:00.000'" + ) + + +def test_iter_rows_accepts_dotted_identifiers(): + """iter_rows accepts fully-qualified catalog.schema.table names.""" + conn = MagicMock() + cursor = _make_cursor(["id"], []) + conn.cursor.return_value = cursor + + # Should not raise + list( + iter_rows( + conn, + "ol_warehouse_production.integrations.integrations__learn__ocw_courses", + ) + ) + + +def test_iter_rows_defers_description_until_first_fetch(): + """Some DB-API drivers only populate cursor.description once results + start arriving, not immediately after execute() (PEP 249 leaves this + driver-defined) — iter_rows must not read it until after the first + fetchmany call. + """ + conn = MagicMock() + cursor = MagicMock() + cursor.description = None + calls = [] + + def _fetchmany(size): + if not calls: + calls.append(1) + cursor.description = [("id",), ("title",)] + return [(1, "Course A")] + return [] + + cursor.fetchmany.side_effect = _fetchmany + conn.cursor.return_value = cursor + + rows = list(iter_rows(conn, "catalog.schema.my_view")) + + assert rows == [{"id": 1, "title": "Course A"}] + + +# --------------------------------------------------------------------------- +# BaseWarehouseETLTask +# --------------------------------------------------------------------------- + + +class _ConcreteTask(BaseWarehouseETLTask): + name = "test.ConcreteTask" + view_name = "ol_warehouse_production.integrations.integrations__learn__test" + + def fetch_and_upsert(self, conn, *, since=None) -> int: # noqa: ARG002 + return 42 + + +class _ErrorTask(BaseWarehouseETLTask): + name = "test.ErrorTask" + view_name = "ol_warehouse_production.integrations.integrations__learn__test" + + def fetch_and_upsert(self, conn, *, since=None) -> int: # noqa: ARG002 + msg = "downstream failure" + raise RuntimeError(msg) + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_base_warehouse_etl_task_run_success(mock_connect): + """run() calls fetch_and_upsert, closes the connection, and returns row count.""" + mock_conn = MagicMock() + mock_connect.return_value = mock_conn + + result = _ConcreteTask().run() + + assert result == 42 + mock_conn.close.assert_called_once() + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_base_warehouse_etl_task_closes_connection_on_error(mock_connect): + """run() closes the warehouse connection even when fetch_and_upsert raises.""" + mock_conn = MagicMock() + mock_connect.return_value = mock_conn + + with pytest.raises(RuntimeError, match="downstream failure"): + _ErrorTask().run() + + mock_conn.close.assert_called_once() + + +@patch("learning_resources.lib.warehouse.sentry_sdk") +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_base_warehouse_etl_task_adds_sentry_breadcrumb_on_error( + mock_connect, mock_sentry +): + """run() adds a Sentry breadcrumb at error level when fetch_and_upsert fails.""" + mock_connect.return_value = MagicMock() + + with pytest.raises(RuntimeError): + _ErrorTask().run() + + mock_sentry.add_breadcrumb.assert_called_once() + call_kwargs = mock_sentry.add_breadcrumb.call_args.kwargs + assert call_kwargs["level"] == "error" + assert call_kwargs["category"] == "warehouse_etl" + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_base_warehouse_etl_task_raises_when_view_name_empty(mock_connect): + """run() raises ValueError immediately when view_name is not set.""" + + class _NoViewTask(BaseWarehouseETLTask): + view_name = "" + + def fetch_and_upsert(self, conn) -> int: # noqa: ARG002 + return 0 + + with pytest.raises(ValueError, match="view_name must be set"): + _NoViewTask().run() + + mock_connect.assert_not_called() + + +def test_base_warehouse_etl_task_fetch_and_upsert_is_abstract(): + """fetch_and_upsert raises NotImplementedError on the base class.""" + task = BaseWarehouseETLTask() + task.view_name = "catalog.schema.view" + with pytest.raises(NotImplementedError): + task.fetch_and_upsert(conn=None) + + +def test_base_warehouse_etl_task_acks_late(): + """acks_late=True, matching the rest of the ETL task fleet (get_ocw_data, + ingest_edx_run_archive, etc.) — a worker lost mid-pull shouldn't lose + the message. + """ + assert BaseWarehouseETLTask.acks_late is True + + +# --------------------------------------------------------------------------- +# full_refresh vs. incremental +# --------------------------------------------------------------------------- + + +class _RecordingTask(BaseWarehouseETLTask): + """Records the `since` it was called with instead of hitting a real view.""" + + name = "test.RecordingTask" + view_name = "ol_warehouse_production.integrations.integrations__learn__test" + + def fetch_and_upsert(self, conn, *, since=None) -> int: # noqa: ARG002 + self.seen_since = since + return 7 + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_full_refresh_defaults_to_true_and_passes_since_none(mock_connect): + """Calling run() with no kwargs is a full refresh: since=None, no watermark I/O.""" + mock_connect.return_value = MagicMock() + task = _RecordingTask() + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + result = task.run() + + assert result == 7 + assert task.seen_since is None + mock_caches.__getitem__.assert_not_called() + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_incremental_reads_watermark_and_passes_it_through(mock_connect): + """full_refresh=False reads the last recorded watermark and forwards it.""" + mock_connect.return_value = MagicMock() + task = _RecordingTask() + stored_watermark = datetime(2026, 6, 1, tzinfo=UTC) + mock_cache = MagicMock() + mock_cache.get.return_value = stored_watermark + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + mock_caches.__getitem__.return_value = mock_cache + result = task.run(full_refresh=False) + + assert result == 7 + assert task.seen_since is stored_watermark + mock_cache.get.assert_called_once_with( + "warehouse_etl:last_synced_at:test.RecordingTask" + ) + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_incremental_with_no_prior_watermark_passes_since_none(mock_connect): + """An incremental run with no recorded watermark yet behaves like a full pull.""" + mock_connect.return_value = MagicMock() + task = _RecordingTask() + mock_cache = MagicMock() + mock_cache.get.return_value = None + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + mock_caches.__getitem__.return_value = mock_cache + task.run(full_refresh=False) + + assert task.seen_since is None + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_incremental_advances_watermark_on_success(mock_connect): + """A successful incremental run stores a fresh watermark, durably.""" + mock_connect.return_value = MagicMock() + task = _RecordingTask() + mock_cache = MagicMock() + mock_cache.get.return_value = None + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + mock_caches.__getitem__.return_value = mock_cache + task.run(full_refresh=False) + + mock_caches.__getitem__.assert_called_with("durable") + mock_cache.set.assert_called_once() + call_args = mock_cache.set.call_args + assert call_args.args[0] == "warehouse_etl:last_synced_at:test.RecordingTask" + assert isinstance(call_args.args[1], datetime) + assert call_args.kwargs["timeout"] is None + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_full_refresh_does_not_advance_watermark(mock_connect): + """A full-refresh run never writes to the watermark cache.""" + mock_connect.return_value = MagicMock() + task = _RecordingTask() + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + task.run(full_refresh=True) + + mock_caches.__getitem__.assert_not_called() + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_incremental_watermark_is_stamped_before_fetch_not_after(mock_connect): + """The watermark records when the fetch *started*, not when it finished. + + Otherwise a row modified while a long-running fetch is in flight would + fall in the gap between this pull's window and the next incremental + run's `since` — invisible until the next full_refresh heals it. + """ + mock_connect.return_value = MagicMock() + fetch_started_at = datetime(2026, 6, 1, 12, 0, 0, tzinfo=UTC) + + with freeze_time(fetch_started_at) as frozen_time: + + class _SlowTask(BaseWarehouseETLTask): + name = "test.SlowTask" + view_name = "ol_warehouse_production.integrations.integrations__learn__test" + + def fetch_and_upsert(self, conn, *, since=None) -> int: # noqa: ARG002 + # Simulate a fetch that takes real wall-clock time. + frozen_time.tick(timedelta(minutes=10)) + return 3 + + task = _SlowTask() + mock_cache = MagicMock() + mock_cache.get.return_value = None + + with patch("learning_resources.lib.warehouse.caches") as mock_caches: + mock_caches.__getitem__.return_value = mock_cache + task.run(full_refresh=False) + + stored_watermark = mock_cache.set.call_args.args[1] + assert stored_watermark == fetch_started_at + + +@patch("learning_resources.lib.warehouse.connect_to_warehouse") +def test_incremental_does_not_advance_watermark_on_failure(mock_connect): + """A failed incremental run leaves the watermark untouched (retry re-covers the gap).""" + mock_connect.return_value = MagicMock() + + class _FailingTask(BaseWarehouseETLTask): + name = "test.FailingTask" + view_name = "ol_warehouse_production.integrations.integrations__learn__test" + + def fetch_and_upsert(self, conn, *, since=None) -> int: # noqa: ARG002 + msg = "boom" + raise RuntimeError(msg) + + task = _FailingTask() + mock_cache = MagicMock() + mock_cache.get.return_value = None + + with ( + patch("learning_resources.lib.warehouse.caches") as mock_caches, + patch("learning_resources.lib.warehouse.sentry_sdk"), + ): + mock_caches.__getitem__.return_value = mock_cache + with pytest.raises(RuntimeError, match="boom"): + task.run(full_refresh=False) + + mock_cache.set.assert_not_called() diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index c5a3524a19..be17f4ff15 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -16,14 +16,16 @@ from learning_resources.constants import LearningResourceType from learning_resources.content_summarizer import ContentSummarizer -from learning_resources.etl import ovs, pipelines, youtube +from learning_resources.etl import catalog_sources, ovs, pipelines, youtube from learning_resources.etl.canvas import ( sync_canvas_archive, ) from learning_resources.etl.constants import ( MARKETING_PAGE_FILE_TYPE, RESOURCE_FILE_ETL_SOURCES, + CourseLoaderConfig, ETLSource, + ProgramLoaderConfig, ) from learning_resources.etl.edx_shared import ( get_most_recent_course_archives, @@ -31,7 +33,9 @@ sync_edx_course_files, ) from learning_resources.etl.loaders import ( + load_courses, load_learning_materials, + load_programs, load_run_dependent_values, ) from learning_resources.etl.pipelines import ocw_courses_etl @@ -39,6 +43,7 @@ get_bucket_by_name, get_s3_prefix_for_source, ) +from learning_resources.lib.warehouse import BaseWarehouseETLTask, iter_rows from learning_resources.models import ContentFile, LearningResource from learning_resources.site_scrapers.utils import scraper_for_site from learning_resources.utils import ( @@ -810,3 +815,192 @@ def cleanup_deleted_content_files(): error = "cleanup_deleted_content_files threw an error" log.exception(error) return error + + +# --------------------------------------------------------------------------- +# Cohort 1: warehouse-pull catalog sources +# +# Each task pulls its integrations__learn__* view from the OL Data +# Platform warehouse (Trino today; see learning_resources.lib.warehouse for +# the backend abstraction), transforms rows with +# learning_resources.etl.catalog_sources, and upserts through the same +# loaders.load_courses/load_programs used by the API-based ETL pipelines +# above. Program tasks fetch_only their child courses, so a program task +# should be scheduled after its courses task has run at least once. +# --------------------------------------------------------------------------- + +# Schema-only, not catalog-qualified: the connection's TRINO_CATALOG +# setting supplies the catalog, so a staging/validation Trino cluster +# with a differently-named catalog is honored via settings alone, +# without these view_name literals drifting out of sync. +_INTEGRATIONS_SCHEMA = "integrations" + + +class SyncMITxOnlineCoursesTask(BaseWarehouseETLTask): + """Warehouse-pull MITx Online courses into MIT Learn.""" + + name = "learning_resources.tasks.SyncMITxOnlineCoursesTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__mitxonline_courses" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + courses = [ + catalog_sources.transform_mitxonline_course(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_courses( + ETLSource.mitxonline.name, + courses, + config=CourseLoaderConfig(prune=since is None), + ) + clear_views_cache() + return len(loaded) + + +SyncMITxOnlineCoursesTask = app.register_task(SyncMITxOnlineCoursesTask()) + + +class SyncMITxOnlineProgramsTask(BaseWarehouseETLTask): + """Warehouse-pull MITx Online programs into MIT Learn.""" + + name = "learning_resources.tasks.SyncMITxOnlineProgramsTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__mitxonline_programs" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + programs = [ + catalog_sources.transform_mitxonline_program(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_programs( + ETLSource.mitxonline.name, + programs, + config=ProgramLoaderConfig( + courses=CourseLoaderConfig(fetch_only=True), prune=since is None + ), + ) + clear_views_cache() + return len(loaded) + + +SyncMITxOnlineProgramsTask = app.register_task(SyncMITxOnlineProgramsTask()) + + +class SyncXProCoursesTask(BaseWarehouseETLTask): + """Warehouse-pull xPRO courses into MIT Learn.""" + + name = "learning_resources.tasks.SyncXProCoursesTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__xpro_courses" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + courses = [ + catalog_sources.transform_xpro_course(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_courses( + ETLSource.xpro.name, courses, config=CourseLoaderConfig(prune=since is None) + ) + clear_views_cache() + return len(loaded) + + +SyncXProCoursesTask = app.register_task(SyncXProCoursesTask()) + + +class SyncXProProgramsTask(BaseWarehouseETLTask): + """Warehouse-pull xPRO programs into MIT Learn.""" + + name = "learning_resources.tasks.SyncXProProgramsTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__xpro_programs" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + programs = [ + catalog_sources.transform_xpro_program(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_programs( + ETLSource.xpro.name, + programs, + config=ProgramLoaderConfig( + courses=CourseLoaderConfig(fetch_only=True), prune=since is None + ), + ) + clear_views_cache() + return len(loaded) + + +SyncXProProgramsTask = app.register_task(SyncXProProgramsTask()) + + +class SyncMITEdXCoursesTask(BaseWarehouseETLTask): + """Warehouse-pull MIT edX (edx.org) courses into MIT Learn.""" + + name = "learning_resources.tasks.SyncMITEdXCoursesTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__mit_edx_courses" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + courses = [ + catalog_sources.transform_mit_edx_course(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_courses( + ETLSource.mit_edx.name, + courses, + config=CourseLoaderConfig(prune=since is None), + ) + clear_views_cache() + return len(loaded) + + +SyncMITEdXCoursesTask = app.register_task(SyncMITEdXCoursesTask()) + + +class SyncOCWCoursesTask(BaseWarehouseETLTask): + """Warehouse-pull OCW courses into MIT Learn.""" + + name = "learning_resources.tasks.SyncOCWCoursesTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__ocw_courses" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + courses = [ + catalog_sources.transform_ocw_course(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_courses( + ETLSource.ocw.name, courses, config=CourseLoaderConfig(prune=since is None) + ) + clear_views_cache() + return len(loaded) + + +SyncOCWCoursesTask = app.register_task(SyncOCWCoursesTask()) + + +class SyncMicromastersProgramsTask(BaseWarehouseETLTask): + """Warehouse-pull MicroMasters programs into MIT Learn.""" + + name = "learning_resources.tasks.SyncMicromastersProgramsTask" + view_name = f"{_INTEGRATIONS_SCHEMA}.integrations__learn__micromasters_programs" + + def fetch_and_upsert(self, conn, *, since=None) -> int: + """Pull, transform, and upsert rows from this task's warehouse view.""" + programs = [ + catalog_sources.transform_micromasters_program(row) + for row in iter_rows(conn, self.view_name, since=since) + ] + loaded = load_programs( + ETLSource.micromasters.name, + programs, + config=ProgramLoaderConfig( + courses=CourseLoaderConfig(fetch_only=True), prune=since is None + ), + ) + clear_views_cache() + return len(loaded) + + +SyncMicromastersProgramsTask = app.register_task(SyncMicromastersProgramsTask()) diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index 0797411892..3e1cf110da 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -1034,3 +1034,178 @@ def test_cleanup_deleted_content_files_returns_error_on_unexpected_exception(moc result = cleanup_deleted_content_files() assert result == "cleanup_deleted_content_files threw an error" + + +@pytest.fixture +def mock_warehouse_connect(mocker): + """Mock the warehouse connection so BaseWarehouseETLTask.run() never dials out.""" + mock_conn = mocker.MagicMock() + mocker.patch( + "learning_resources.lib.warehouse.connect_to_warehouse", return_value=mock_conn + ) + return mock_conn + + +@pytest.mark.parametrize( + ("task_name", "view_suffix", "transform_name", "etl_source"), + [ + ( + "SyncMITxOnlineCoursesTask", + "integrations__learn__mitxonline_courses", + "transform_mitxonline_course", + ETLSource.mitxonline.name, + ), + ( + "SyncXProCoursesTask", + "integrations__learn__xpro_courses", + "transform_xpro_course", + ETLSource.xpro.name, + ), + ( + "SyncMITEdXCoursesTask", + "integrations__learn__mit_edx_courses", + "transform_mit_edx_course", + ETLSource.mit_edx.name, + ), + ( + "SyncOCWCoursesTask", + "integrations__learn__ocw_courses", + "transform_ocw_course", + ETLSource.ocw.name, + ), + ], +) +def test_warehouse_sync_courses_task_wiring( # noqa: PLR0913 + mocker, mock_warehouse_connect, task_name, view_suffix, transform_name, etl_source +): + """Each Cohort 1 courses task pulls its view, transforms rows, and loads courses.""" + task = getattr(tasks, task_name) + rows = [{"readable_id": "a"}, {"readable_id": "b"}] + mock_iter_rows = mocker.patch( + "learning_resources.tasks.iter_rows", return_value=rows + ) + mock_transform = mocker.patch( + f"learning_resources.tasks.catalog_sources.{transform_name}" + ) + loaded = LearningResourceFactory.create_batch(2) + mock_load_courses = mocker.patch( + "learning_resources.tasks.load_courses", return_value=loaded + ) + mock_clear_cache = mocker.patch("learning_resources.tasks.clear_views_cache") + + result = task.delay().get() + + assert result == len(loaded) + assert view_suffix in mock_iter_rows.call_args.args[1] + assert mock_transform.call_count == len(rows) + mock_load_courses.assert_called_once() + assert mock_load_courses.call_args.args[0] == etl_source + mock_clear_cache.assert_called_once() + mock_warehouse_connect.close.assert_called_once() + + +@pytest.mark.parametrize( + ("task_name", "view_suffix", "transform_name", "etl_source"), + [ + ( + "SyncMITxOnlineProgramsTask", + "integrations__learn__mitxonline_programs", + "transform_mitxonline_program", + ETLSource.mitxonline.name, + ), + ( + "SyncXProProgramsTask", + "integrations__learn__xpro_programs", + "transform_xpro_program", + ETLSource.xpro.name, + ), + ( + "SyncMicromastersProgramsTask", + "integrations__learn__micromasters_programs", + "transform_micromasters_program", + ETLSource.micromasters.name, + ), + ], +) +def test_warehouse_sync_programs_task_wiring( # noqa: PLR0913 + mocker, mock_warehouse_connect, task_name, view_suffix, transform_name, etl_source +): + """Each Cohort 1 programs task pulls its view, transforms rows, and loads programs.""" + task = getattr(tasks, task_name) + rows = [{"readable_id": "a"}] + mock_iter_rows = mocker.patch( + "learning_resources.tasks.iter_rows", return_value=rows + ) + mock_transform = mocker.patch( + f"learning_resources.tasks.catalog_sources.{transform_name}" + ) + loaded = LearningResourceFactory.create_batch(1) + mock_load_programs = mocker.patch( + "learning_resources.tasks.load_programs", return_value=loaded + ) + mock_clear_cache = mocker.patch("learning_resources.tasks.clear_views_cache") + + result = task.delay().get() + + assert result == len(loaded) + assert view_suffix in mock_iter_rows.call_args.args[1] + assert mock_transform.call_count == len(rows) + mock_load_programs.assert_called_once() + assert mock_load_programs.call_args.args[0] == etl_source + mock_clear_cache.assert_called_once() + mock_warehouse_connect.close.assert_called_once() + + +def test_warehouse_sync_task_closes_connection_on_error(mocker, mock_warehouse_connect): + """A warehouse-pull task closes the connection even when the fetch/transform fails.""" + mocker.patch( + "learning_resources.tasks.iter_rows", + side_effect=RuntimeError("trino query failed"), + ) + + with pytest.raises(RuntimeError, match="trino query failed"): + tasks.SyncOCWCoursesTask.delay().get() + + mock_warehouse_connect.close.assert_called_once() + + +def test_warehouse_sync_task_full_refresh_pulls_everything_and_prunes( + mocker, mock_warehouse_connect +): + """The default full_refresh=True run pulls unfiltered and prunes stale resources.""" + mock_iter_rows = mocker.patch("learning_resources.tasks.iter_rows", return_value=[]) + mock_load_courses = mocker.patch( + "learning_resources.tasks.load_courses", return_value=[] + ) + mocker.patch("learning_resources.tasks.clear_views_cache") + mock_caches = mocker.patch("learning_resources.lib.warehouse.caches") + + tasks.SyncOCWCoursesTask.delay().get() + + assert mock_iter_rows.call_args.kwargs["since"] is None + assert mock_load_courses.call_args.kwargs["config"].prune is True + mock_caches.__getitem__.assert_not_called() + + +def test_warehouse_sync_task_incremental_skips_prune_and_advances_watermark( + mocker, mock_warehouse_connect +): + """full_refresh=False threads the stored watermark through and skips pruning.""" + stored_watermark = now_in_utc() - timedelta(days=1) + mock_iter_rows = mocker.patch("learning_resources.tasks.iter_rows", return_value=[]) + mock_load_courses = mocker.patch( + "learning_resources.tasks.load_courses", return_value=[] + ) + mocker.patch("learning_resources.tasks.clear_views_cache") + mock_cache = mocker.MagicMock() + mock_cache.get.return_value = stored_watermark + mocker.patch("learning_resources.lib.warehouse.caches", {"durable": mock_cache}) + + tasks.SyncOCWCoursesTask.apply_async(kwargs={"full_refresh": False}).get() + + assert mock_iter_rows.call_args.kwargs["since"] is stored_watermark + assert mock_load_courses.call_args.kwargs["config"].prune is False + mock_cache.set.assert_called_once() + assert mock_cache.set.call_args.args[0] == ( + "warehouse_etl:last_synced_at:learning_resources.tasks.SyncOCWCoursesTask" + ) diff --git a/main/settings_celery.py b/main/settings_celery.py index 883781892e..a20aae564c 100644 --- a/main/settings_celery.py +++ b/main/settings_celery.py @@ -205,6 +205,66 @@ } ) +# Cohort 1 warehouse-pull catalog sources (learning_resources.tasks.Sync*Task). +# Scheduled to run in parallel with the existing API-based ETL tasks above +# during validation; see implementation_guide_01_db_catalog.md. Programs run +# 15 minutes after their courses task so fetch_only child course lookups find +# freshly-synced course resources. +# +# All entries below run full_refresh=True (the safe, pruning-capable +# baseline). Each task also supports full_refresh=False for a cheaper, +# watermark-based incremental pull — see +# learning_resources.lib.warehouse.BaseWarehouseETLTask — for a future +# higher-frequency schedule tier once Cohort 1 is validated; not wired up +# here yet. +# +# Gated on TRINO_HOST being configured: read directly here (not imported +# from settings_course_etl, which loads after settings_celery — see +# main/settings.py's import order) so these entries are simply absent, not +# registered-but-guaranteed-to-fail, in any environment without warehouse +# connectivity (no environment has it yet; see +# https://github.com/mitodl/hq/issues/11509). +if not CELERY_BEAT_DISABLED and get_string("TRINO_HOST", None): + CELERY_BEAT_SCHEDULE.update( + { + "warehouse-sync-mitxonline-courses-every-1-days": { + "task": "learning_resources.tasks.SyncMITxOnlineCoursesTask", + "schedule": crontab(minute=0, hour=10), # 6:00am EDT / 5:00am EST + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-mitxonline-programs-every-1-days": { + "task": "learning_resources.tasks.SyncMITxOnlineProgramsTask", + "schedule": crontab(minute=15, hour=10), + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-xpro-courses-every-1-days": { + "task": "learning_resources.tasks.SyncXProCoursesTask", + "schedule": crontab(minute=30, hour=10), + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-xpro-programs-every-1-days": { + "task": "learning_resources.tasks.SyncXProProgramsTask", + "schedule": crontab(minute=45, hour=10), + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-mit-edx-courses-every-1-days": { + "task": "learning_resources.tasks.SyncMITEdXCoursesTask", + "schedule": crontab(minute=0, hour=11), + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-ocw-courses-every-1-days": { + "task": "learning_resources.tasks.SyncOCWCoursesTask", + "schedule": crontab(minute=15, hour=11), + "kwargs": {"full_refresh": True}, + }, + "warehouse-sync-micromasters-programs-every-1-days": { + "task": "learning_resources.tasks.SyncMicromastersProgramsTask", + "schedule": crontab(minute=30, hour=11), + "kwargs": {"full_refresh": True}, + }, + } + ) + CELERY_TASK_SERIALIZER = "json" CELERY_RESULT_SERIALIZER = "json" CELERY_ACCEPT_CONTENT = ["json"] diff --git a/main/settings_course_etl.py b/main/settings_course_etl.py index 96ffcfa36a..d29d59480a 100644 --- a/main/settings_course_etl.py +++ b/main/settings_course_etl.py @@ -145,3 +145,16 @@ "CONTENT_BASE_URL_OLL", "https://openlearninglibrary.mit.edu" ) CONTENT_BASE_URL_EDX = get_string("CONTENT_BASE_URL_EDX", "https://courses.edx.org") + +# Warehouse-pull settings for Cohort 1 catalog ETL, used to query the +# integrations schema views exposed by the OL Data Platform. The query +# engine backing this endpoint may migrate from Trino to StarRocks later; +# WAREHOUSE_BACKEND selects the connector without touching ETL task code +# (see learning_resources.lib.warehouse). Backend-specific credentials are +# namespaced (e.g. TRINO_*) so a future STARROCKS_* block can sit alongside. +WAREHOUSE_BACKEND = get_string("WAREHOUSE_BACKEND", "trino") +TRINO_HOST = get_string("TRINO_HOST", None) +TRINO_PORT = get_int("TRINO_PORT", 443) +TRINO_USER = get_string("TRINO_USER", None) +TRINO_PASSWORD = get_string("TRINO_PASSWORD", None) +TRINO_CATALOG = get_string("TRINO_CATALOG", "ol_warehouse_production") diff --git a/pyproject.toml b/pyproject.toml index bbdc431c71..7cc59e6d2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -118,6 +118,7 @@ dependencies = [ "gensim>=4.4.0", "langchain-community>=0.4.2", "langchain-text-splitters>=1.1.2", + "trino>=0.334.0", ] [dependency-groups] diff --git a/uv.lock b/uv.lock index 1bed38aa9e..a8efee9e4b 100644 --- a/uv.lock +++ b/uv.lock @@ -2457,6 +2457,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/92/96/a5dc078cf0126fbfbc35611d77ecd5da80054b5893e28fb213a5613b9e1d/lxml-6.1.0-cp312-cp312-win_arm64.whl", hash = "sha256:c3592631e652afa34999a088f98ba7dfc7d6aff0d535c410bea77a71743f3819", size = 3659552, upload-time = "2026-04-18T04:27:51.133Z" }, ] +[[package]] +name = "lz4" +version = "4.4.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/57/51/f1b86d93029f418033dddf9b9f79c8d2641e7454080478ee2aab5123173e/lz4-4.4.5.tar.gz", hash = "sha256:5f0b9e53c1e82e88c10d7c180069363980136b9d7a8306c4dca4f760d60c39f0", size = 172886, upload-time = "2025-11-03T13:02:36.061Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1b/ac/016e4f6de37d806f7cc8f13add0a46c9a7cfc41a5ddc2bc831d7954cf1ce/lz4-4.4.5-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:df5aa4cead2044bab83e0ebae56e0944cc7fcc1505c7787e9e1057d6d549897e", size = 207163, upload-time = "2025-11-03T13:01:45.895Z" }, + { url = "https://files.pythonhosted.org/packages/8d/df/0fadac6e5bd31b6f34a1a8dbd4db6a7606e70715387c27368586455b7fc9/lz4-4.4.5-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6d0bf51e7745484d2092b3a51ae6eb58c3bd3ce0300cf2b2c14f76c536d5697a", size = 207150, upload-time = "2025-11-03T13:01:47.205Z" }, + { url = "https://files.pythonhosted.org/packages/b7/17/34e36cc49bb16ca73fb57fbd4c5eaa61760c6b64bce91fcb4e0f4a97f852/lz4-4.4.5-cp312-cp312-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:7b62f94b523c251cf32aa4ab555f14d39bd1a9df385b72443fd76d7c7fb051f5", size = 1292045, upload-time = "2025-11-03T13:01:48.667Z" }, + { url = "https://files.pythonhosted.org/packages/90/1c/b1d8e3741e9fc89ed3b5f7ef5f22586c07ed6bb04e8343c2e98f0fa7ff04/lz4-4.4.5-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2c3ea562c3af274264444819ae9b14dbbf1ab070aff214a05e97db6896c7597e", size = 1279546, upload-time = "2025-11-03T13:01:50.159Z" }, + { url = "https://files.pythonhosted.org/packages/55/d9/e3867222474f6c1b76e89f3bd914595af69f55bf2c1866e984c548afdc15/lz4-4.4.5-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:24092635f47538b392c4eaeff14c7270d2c8e806bf4be2a6446a378591c5e69e", size = 1368249, upload-time = "2025-11-03T13:01:51.273Z" }, + { url = "https://files.pythonhosted.org/packages/b2/e7/d667d337367686311c38b580d1ca3d5a23a6617e129f26becd4f5dc458df/lz4-4.4.5-cp312-cp312-win32.whl", hash = "sha256:214e37cfe270948ea7eb777229e211c601a3e0875541c1035ab408fbceaddf50", size = 88189, upload-time = "2025-11-03T13:01:52.605Z" }, + { url = "https://files.pythonhosted.org/packages/a5/0b/a54cd7406995ab097fceb907c7eb13a6ddd49e0b231e448f1a81a50af65c/lz4-4.4.5-cp312-cp312-win_amd64.whl", hash = "sha256:713a777de88a73425cf08eb11f742cd2c98628e79a8673d6a52e3c5f0c116f33", size = 99497, upload-time = "2025-11-03T13:01:53.477Z" }, + { url = "https://files.pythonhosted.org/packages/6a/7e/dc28a952e4bfa32ca16fa2eb026e7a6ce5d1411fcd5986cd08c74ec187b9/lz4-4.4.5-cp312-cp312-win_arm64.whl", hash = "sha256:a88cbb729cc333334ccfb52f070463c21560fca63afcf636a9f160a55fac3301", size = 91279, upload-time = "2025-11-03T13:01:54.419Z" }, +] + [[package]] name = "markdown" version = "3.10.2" @@ -2647,6 +2663,7 @@ dependencies = [ { name = "tiktoken" }, { name = "tldextract" }, { name = "toolz" }, + { name = "trino" }, { name = "ulid-py" }, { name = "urllib3" }, { name = "wrapt" }, @@ -2787,6 +2804,7 @@ requires-dist = [ { name = "tiktoken", specifier = ">=0.12.0,<0.13" }, { name = "tldextract", specifier = ">=5.0.0,<6" }, { name = "toolz", specifier = ">=1.0.0,<2" }, + { name = "trino", specifier = ">=0.334.0" }, { name = "ulid-py", specifier = ">=1.0.0,<2" }, { name = "urllib3", specifier = ">=2.0.0,<3" }, { name = "wrapt", specifier = ">=1.14.1,<2" }, @@ -5076,6 +5094,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/00/c0/8f5d070730d7836adc9c9b6408dec68c6ced86b304a9b26a14df072a6e8c/traitlets-5.14.3-py3-none-any.whl", hash = "sha256:b74e89e397b1ed28cc831db7aea759ba6640cb3de13090ca145426688ff1ac4f", size = 85359, upload-time = "2024-04-19T11:11:46.763Z" }, ] +[[package]] +name = "trino" +version = "0.338.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "lz4" }, + { name = "orjson", marker = "platform_python_implementation != 'PyPy'" }, + { name = "python-dateutil" }, + { name = "pytz" }, + { name = "requests" }, + { name = "tzlocal" }, + { name = "zstandard" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/18/aa/4bc0539467c12d415e8f8870a79646be80d299c063c563a47fee0b6c9fa9/trino-0.338.0.tar.gz", hash = "sha256:1103d249cb96ba8f88f699588581965e2be088c16b9f8178e34cb64aa8a6c011", size = 57373, upload-time = "2026-06-29T19:55:37.547Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/91/b5/182b81d1b4cc2b306b174bd94efe5384f7d1253539eaf47a6b4d842b96a7/trino-0.338.0-py3-none-any.whl", hash = "sha256:093952f8a53b6f47ab357b77cbe92f1d66453290f7d8bcc4b74bd662e5ff2704", size = 59981, upload-time = "2026-06-29T19:55:36.449Z" }, +] + [[package]] name = "trio" version = "0.33.0"