feat(cohort-1): warehouse-pull Celery tasks for catalog sources#3566
feat(cohort-1): warehouse-pull Celery tasks for catalog sources#3566blarghmatey wants to merge 4 commits into
Conversation
Implements the Cohort 1 Trino-pull ETL pattern: 7 Sync*Task Celery tasks (MITx Online courses/programs, xPRO courses/programs, MIT edX courses, OCW courses, MicroMasters programs) that pull integrations__learn__* views from the OL Data Platform warehouse and upsert through the existing loaders.load_courses/load_programs pipeline. - learning_resources/lib/warehouse.py: backend-agnostic warehouse-pull infrastructure (BaseWarehouseETLTask, iter_rows) built on the DB-API 2.0 cursor surface so a future StarRocks/DuckDB backend is a drop-in connector, not an ETL-layer rewrite. Trino is the only wired-up backend today, selected via settings.WAREHOUSE_BACKEND. - learning_resources/etl/catalog_sources.py: row transforms mapping each integrations__learn__* view's flattened contract to the course/program dict shape the existing loaders expect. - Sync tasks support full_refresh (default, prunes stale resources) vs. incremental (since=<last watermark>, skips pruning) modes, selected via a full_refresh kwarg. Watermarks persist in the durable DB-backed cache so an incremental run resumes correctly across process restarts. - Beat schedule entries registered in main/settings_celery.py, running full_refresh=True daily alongside the existing API-based ETL tasks during Cohort 1 parallel validation. Also fixes a latent bug in the view-name safety check: the regex used `$` instead of `\Z`, so a view_name ending in a trailing newline slipped past the "unsafe identifier" guard (Python's `$` matches just before a trailing newline, not strictly end-of-string).
OpenAPI ChangesNo changes detected Unexpected changes? Ensure your branch is up-to-date with |
for more information, see https://pre-commit.ci
There was a problem hiding this comment.
Pull request overview
This PR introduces a “warehouse-pull” ETL pattern for Cohort 1 catalog sources, adding new Celery tasks that query pre-computed integrations__learn__* views (Trino today) and upsert them via the existing course/program loaders. It also adds a small SQL-injection guard fix and supporting transforms/tests to enable parallel validation against the current API-based ETL pipelines.
Changes:
- Added warehouse connectivity + row-iteration infrastructure (
BaseWarehouseETLTask,connect_to_warehouse,iter_rows) with full-refresh vs incremental (watermark-based) support. - Added 7 new Celery tasks to sync catalog sources from warehouse views and scheduled them in Celery beat for daily full refresh.
- Added
catalog_sourcesrow→loader-shape transforms and comprehensive unit tests for tasks, transforms, and warehouse helpers.
Reviewed changes
Copilot reviewed 10 out of 12 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Locks trino and its dependencies (lz4, etc.) for the new warehouse connector. |
| pyproject.toml | Adds trino>=0.334.0 dependency for warehouse connectivity. |
| main/settings_course_etl.py | Introduces warehouse backend + Trino connection settings. |
| main/settings_celery.py | Adds daily Celery beat schedules for the new warehouse-sync tasks. |
| learning_resources/tasks.py | Implements 7 new warehouse-pull ETL Celery tasks for Cohort 1 catalog sources. |
| learning_resources/tasks_test.py | Adds integration-style tests validating task wiring, full vs incremental behavior, and connection cleanup. |
| learning_resources/lib/warehouse.py | Adds backend-agnostic warehouse connection dispatch, safe view iteration, and the BaseWarehouseETLTask base class. |
| learning_resources/lib/warehouse_test.py | Adds fast unit tests for connection dispatch, iter_rows, and watermark logic without loading the full Django app. |
| learning_resources/lib/init.py | Ensures learning_resources.lib is a package for the new warehouse module. |
| learning_resources/etl/catalog_sources.py | Adds dict→dict transforms mapping flattened warehouse rows into the loader input shape. |
| learning_resources/etl/catalog_sources_test.py | Adds unit tests for transform helpers and per-source transforms. |
| env/backend.local.example.env | Documents new warehouse-related environment variables for local setup. |
- _connect_trino: fail fast with ImproperlyConfigured when TRINO_HOST/ TRINO_USER are unset, instead of constructing BasicAuthentication(None, None) and surfacing a confusing low-level connect() failure. Auth is now optional (None) when TRINO_USER/TRINO_PASSWORD aren't both set, rather than always wrapping possibly-None credentials. - tasks.py: stop hardcoding the "ol_warehouse_production" catalog into every Sync*Task's view_name. Confirmed via the installed trino client (trino.dbapi.Connection accepts catalog=... and threads it through ClientSession, so 2-part schema.table references resolve against the connection's default catalog) that view_name only needs to be schema-qualified — settings.TRINO_CATALOG already supplies the catalog on the connection. A staging/validation Trino cluster with a differently-named catalog is now honored via settings alone. - settings_celery.py: gate the 7 warehouse-sync beat entries behind TRINO_HOST being configured (read directly, since settings_celery loads before settings_course_etl per main/settings.py's import order). No environment has Trino network access yet (mitodl/hq#11509), so these entries would otherwise be registered-but-guaranteed-to-fail, paging on connection errors daily. Fixed the accompanying schedule comment: hour=10 UTC is 6am EDT / 5am EST, not "6:00am EST" outright. - catalog_sources.py: _parse_datetime now normalizes to UTC via parse(value).replace(tzinfo=UTC), matching the identical convention in learning_resources/etl/mitxonline.py and xpro.py — naive datetimes were triggering Django's naive-datetime warnings and were ambiguous under server-timezone assumptions. _split's default separator changed from ", " to "," (still .strip()'d per-part) to match the plain-comma convention used elsewhere in learning_resources/etl (ocw.py, podcast.py, sloan.py) — verified this is a no-op for the actual ol-data-platform views, which array_join on ", ", since strip() normalizes either separator to the same result, but plain "," is also correct if a view ever emits unspaced commas. All 116 tests in the touched files still pass; ruff check/format clean (3 remaining D103 findings in tasks.py are pre-existing, unrelated to this PR). Verified via `manage.py check` and a direct settings import that the beat-schedule gate actually produces 0 warehouse-sync entries without TRINO_HOST set and 7 with it set.
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
Sentry flagged that transform_micromasters_program omits "topics"
entirely, and loaders.load_program's `program_data.pop("topics", [])`
defaults to [] for a missing key — which load_topics treats as "clear
all topics", wiping any existing MicroMasters program topics on every
sync.
integrations__learn__micromasters_programs has no topics column (unlike
every other Cohort 1 view), so there's nothing to populate it with yet.
Set "topics": None instead, which is loaders.py's own existing sentinel
for "not provided, leave alone" (load_topics only acts when
`topics_data is not None`; the same None-vs-[] distinction already
governs `departments_data = program_data.pop("departments", None)` a few
lines below). This stops the data loss without inventing new loader
semantics.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code review is billed via overage credits. To resume reviews, an organization admin can raise the monthly limit at claude.ai/admin-settings/claude-code.
Once credits are available, push a new commit or reopen this pull request to trigger a review.
| "Susskind": PlatformType.susskind.name, | ||
| "WHU": PlatformType.whu.name, | ||
| "xPRO": PlatformType.xpro.name, | ||
| } |
There was a problem hiding this comment.
Move to constants.py alongside other dicts?
| "start_date": _parse_datetime(start_on), | ||
| "end_date": _parse_datetime(end_on), | ||
| "instructors": instructors, | ||
| "prices": [], |
There was a problem hiding this comment.
load_run unconditionally overwrites run.prices (loaders.py:327-328) and load_prices(run, []) clears resource_prices — omitting the key doesn't help. Since these tasks upsert the same rows as the API ETL (same etl_source + readable_id), every warehouse sync will wipe displayed prices for mitxonline/xpro/mit_edx runs until the next API run restores them — a daily, user-visible flap once the beat schedule goes live. Consider making parallel validation read-only against prod rows, or adding prices to the views first.
There was a problem hiding this comment.
#3565 is aimed at handling this shift in responsibilities as we cut over different data flows.
| """ | ||
| title = row["title"] | ||
| return { | ||
| "readable_id": row["readable_id"], |
There was a problem hiding this comment.
The view emits cast(program_id as varchar) ("3", "4"), but the API ETL synthesizes micromasters-program-{id} (micromasters.py:126) — prod currently has micromasters-program-3/4 (verified via the public API). Passing the bare id through creates duplicate programs, and the prune step then unpublishes the real ones; the next API run reverses it. Prefix here (and use the same prefixed value for run_id, which the API ETL also prefixes — micromasters.py:137), or fix the view in ol-data-platform.
| conn.close() | ||
|
|
||
| if not full_refresh: | ||
| self._set_watermark(datetime.now(tz=UTC)) |
There was a problem hiding this comment.
The watermark is stamped after the fetch completes, so rows modified between query execution and completion fall outside both this pull and the next incremental window (skipped until a full refresh heals them). Capture the timestamp before executing the query, or use the max last_modified seen. Not urgent while incremental is unscheduled, but worth fixing before that tier goes live.
| return [part.strip() for part in value.split(sep) if part.strip()] | ||
|
|
||
|
|
||
| def _parse_bool(value) -> bool: |
There was a problem hiding this comment.
_parse_bool(1) returns False. Trino returns real bools, but StarRocks (MySQL protocol) returns 1/0 — the migration this module explicitly plans for — which would silently unpublish everything. Suggest: if isinstance(value, (bool, int)): return bool(value).
| "published": published, | ||
| "url": row.get("url"), | ||
| "semester": row.get("term"), | ||
| "year": row.get("year"), |
There was a problem hiding this comment.
year lands in an IntegerField — "2025" coerces fine, but an empty string would raise on save. Cheap to guard here, e.g. int(row["year"]) if row.get("year") else None.
| cur.close() | ||
|
|
||
|
|
||
| class BaseWarehouseETLTask(Task): |
There was a problem hiding this comment.
BaseWarehouseETLTask doesn't set acks_late=True. The other long-running ETL tasks in this module do (e.g. get_ocw_data, ingest_edx_run_archive, import_*_files). A worker lost mid-pull loses the work and the message is already acked. Low-impact today (daily full_refresh=True is idempotent/self-healing, and a failed incremental leaves the watermark untouched so the next run re-covers the gap), but acks_late=True on the base would match the rest of the ETL task fleet. Optional.
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| def transform_micromasters_program(row: dict) -> dict: |
There was a problem hiding this comment.
Distinct from the prices-wipe and readable_id issues already flagged: this transform also omits certification, certification_type, start_date/end_date, availability, and pace, all of which the API ETL sets (micromasters.py:133-149). Those are non-destructive — loaders only write keys present in the dict — but post-cutover they'd freeze at their last API-written values. Worth recording in the cutover plan: extend the view with these columns, or explicitly keep the API path as owner of them.
instructors is the exception, and it's destructive: _program_run provides no instructors key, so load_run pops the default [] and load_instructors(run, []) deletes every RunInstructorRelationship for the run (loaders.py:224). Program-run instructors will be wiped on every warehouse sync — same class of issue as prices, and this applies to the mitxonline/xpro program runs too.
What are the relevant tickets?
Part of https://github.com/mitodl/hq/issues/11510 (Cohort 1: DB-Backed Catalog Sources / Trino-pull)
Description (What does it do?)
Implements the Cohort 1 Trino-pull ETL pattern for MIT Learn's OL Data Platform migration: 7 new Celery tasks that pull pre-computed
integrations__learn__*views from the warehouse and upsert them through the existing course/program loaders, replacing (once cut over) the equivalent API-based ETL pipelines for these sources.New tasks (
learning_resources/tasks.py):SyncMITxOnlineCoursesTask,SyncMITxOnlineProgramsTask,SyncXProCoursesTask,SyncXProProgramsTask,SyncMITEdXCoursesTask,SyncOCWCoursesTask,SyncMicromastersProgramsTask. Program tasksfetch_onlytheir child courses, so are scheduled 15 minutes after their courses task.New infrastructure (
learning_resources/lib/warehouse.py):BaseWarehouseETLTask— a Celery task base class that opens a warehouse connection, delegates tofetch_and_upsert, and handles connection lifecycle/error reporting (Sentry breadcrumbs).iter_rows()— backend-agnostic row iteration built on the plain DB-API 2.0 cursor surface (execute/description/fetchmany). Trino is the only backend wired up today (settings.WAREHOUSE_BACKEND); a_connect_starrocksstub documents the intended migration path — since Trino, StarRocks, and DuckDB all expose the same cursor surface, adding a backend is a connector fill-in, not an ETL-layer rewrite.run()takes afull_refresh: bool = Truekwarg.full_refresh=True(today's beat schedule default) pulls every row and prunes resources no longer present in the source — the self-healing baseline.full_refresh=Falsepulls only rows withlast_modifiedgreater than a persisted watermark (stored in Django's existing durable DB-backed cache, not a new model/migration) and skips pruning, since a partial pull must never be treated as the complete source state. Everyintegrations__learn__*view already exposeslast_modifiedper the platform's marts contract, so this required no changes on the data platform side.New transforms (
learning_resources/etl/catalog_sources.py): puredict -> dictfunctions mapping each view's flattened row contract (delimited strings for topics/instructors/runs, since the views intentionally avoid nested JSON) to the shapeloaders.load_courses/load_programsexpect.Bug fix:
_SAFE_IDENTIFIER's regex used$instead of\Zto anchor the "safe view name" check. Python's$matches just before a trailing newline (not strictly end-of-string), so aview_nameending in\nslipped past the guard. Fixed to\Z.How can this be tested?
No live Trino endpoint is available yet in any environment (see https://github.com/mitodl/hq/issues/11509, "confirm MIT Learn deployment environment has network access to the Trino/Starburst endpoint" — still open), so there is no live-connectivity smoke test in this PR. All 116 tests in the touched files pass against mocked warehouse connections:
warehouse_test.pycovers connection dispatch (Trino/StarRocks/unknown backend),iter_rowsbatching/cursor-cleanup/SQL-injection-guard/since-filtering, andBaseWarehouseETLTask's full-refresh vs. incremental branching (watermark read/write, prune flag, error paths) — all against a hand-built DB-API-2.0-shaped mock, no Django app/DB required.tasks_test.pycovers each of the 7 tasks' wiring (view name -> transform -> loader ->clear_views_cache) plus explicit full-refresh vs. incremental coverage, against a real Django test DB.catalog_sources_test.pycovers the row-transform functions directly.Reviewers can also sanity-check the beat schedule wiring in
main/settings_celery.pyand confirm thefull_refresh=Truekwargs match the intended daily-parallel-validation cadence described in the cohort's implementation guide.Additional Context
Opening as a draft: this is Cohort 1 infrastructure landing ahead of the still-open network-access confirmation (https://github.com/mitodl/hq/issues/11509) and before parallel validation against the existing API-based ETL has started. Once Trino connectivity is confirmed from this deployment environment, the plan is to run both ETL paths in parallel for 2 weeks (≥99% record match) before cutover — tracked as a separate task in the epic.