feat(dlt): add a reusable database source, with Keycloak as its first instance

dg_projects/data_loading/data_loading/defs/ingestion/assets.py

@@ -14,6 +14,7 @@
 from dagster_dlt import DagsterDltResource, dlt_assets
 from ol_dlt.sources import (
     edxorg_s3,
+    keycloak,
     mit_climate,
     mit_edx_programs,
     mitpe,
@@ -73,6 +74,11 @@ def _assets(
     source=podcast_rss.build_source(),
     pipeline=podcast_rss.podcast_rss_pipeline,
 )
+keycloak_assets = build_ingest_assets(
+    name="keycloak_ingest",
+    source=keycloak.build_source(),
+    pipeline=keycloak.keycloak_pipeline,
+)
 
 
 # --- edxorg_s3: custom upstream deps + one op per table ---------------------
@@ -131,6 +137,7 @@ def _asset(
         mit_climate_assets,
         mit_edx_programs_assets,
         podcast_rss_assets,
+        keycloak_assets,
         *edxorg_s3_table_assets,
     ],
 )

dg_projects/data_loading/data_loading/defs/ingestion/schedules.py

@@ -48,12 +48,22 @@
     execution_timezone="Etc/UTC",
 )
 
+keycloak_ingest_schedule = dg.ScheduleDefinition(
+    name="keycloak_ingest_daily_schedule",
+    # Selected by group rather than by key so adding a table to KEYCLOAK_SPEC
+    # does not also require editing this schedule.
+    target=dg.AssetSelection.groups("keycloak"),
+    cron_schedule="30 4 * * *",
+    execution_timezone="Etc/UTC",
+)
+
 defs = dg.Definitions(
     schedules=[
         oll_ingest_schedule,
         mitpe_ingest_schedule,
         mit_climate_ingest_schedule,
         mit_edx_programs_ingest_schedule,
         podcast_rss_ingest_schedule,
+        keycloak_ingest_schedule,
     ],
 )

src/ol_dlt/README.md

@@ -21,6 +21,8 @@ src/ol_dlt/
 │   └── secrets.toml.template
 ├── ol_dlt/
 │   ├── config.py         # profile -> destination/dataset/table_format factory
+│   ├── database.py       # reusable relational-database source (spec -> @dlt.source)
+│   ├── vault.py          # Vault-issued database credentials (pure hvac)
 │   └── sources/<name>/   # @dlt.source / @dlt.resource bodies only
 └── tests/                # unit + materialization tests (ephemeral DuckDB/filesystem)
 ```
@@ -96,6 +98,8 @@ DLT_PROFILE=dev uv run python -m ol_dlt.sources.podcast_rss
 DLT_PROFILE=dev uv run python -m ol_dlt.sources.mit_edx_programs
 #    Needs AWS creds (reads the prod S3 landing zone):
 DLT_PROFILE=dev uv run python -m ol_dlt.sources.edxorg_s3
+#    Needs the local-dev Postgres cluster port-forwarded (see Database sources):
+DLT_PROFILE=dev uv run python -m ol_dlt.sources.keycloak
 
 # 5. Inspect a pipeline's state / last load after a run.
 DLT_PROFILE=dev uv run dlt pipeline oll info
@@ -104,10 +108,53 @@ DLT_PROFILE=dev uv run dlt pipeline oll info
 The hermetic suite (step 1) is what CI runs and what should gate merges; the
 live runs (step 4) depend on the upstream services being reachable.
 
+## Database sources
+
+Relational sources do **not** get hand-written resources. Declare a
+`DatabaseSourceSpec` in `ol_dlt/sources/<name>/__init__.py` and
+`ol_dlt.database` builds the `@dlt.source` from it — see
+`ol_dlt/sources/keycloak/` for the reference instance.
+
+```python
+KEYCLOAK_SPEC = DatabaseSourceSpec(
+    name="keycloak",
+    raw_table_prefix="raw__keycloak__app__postgres__",
+    database="keycloak",
+    vault_mount="postgres-keycloak",
+    tables=(DatabaseTable(name="user_entity", primary_key="id"), ...),
+)
+```
+
+**Which database gets read is the profile's job, not the spec's.** `qa` reads
+the QA database, `production` reads production, and `dev`/`ci`/`test` read the
+local-dev CloudNativePG cluster from `ol-infrastructure/local-dev`, so no
+developer needs access to a deployed environment to exercise a pipeline:
+
+```bash
+kubectl port-forward -n local-infra svc/local-pg-rw 5432:5432
+DLT_PROFILE=dev uv run python -m ol_dlt.sources.keycloak
+```
+
+Per-source overrides come from `<NAME>_DB_HOST` / `_PORT` / `_USERNAME` /
+`_PASSWORD`. Deployed profiles **require** `<NAME>_DB_HOST` (the Dagster Pulumi
+stack passes it through from the application's stack reference) and take their
+username/password from Vault at connection time — never from the environment,
+and never at code-location import, where a lease would go stale long before a
+scheduled run fires.
+
+Two things that need doing outside this repo before a new database source can
+run in a deployed environment, both in `ol-infrastructure`:
+
+1. Grant `<mount>/creds/readonly` in
+   `src/ol_infrastructure/applications/dagster/dagster_server_policy.hcl`.
+2. Add `<NAME>_DB_HOST` to the `data_loading` deployment env in the dagster
+   `__main__.py`, sourced from the application stack's exported RDS host.
+
 ## Adding a new source
 
 1. Author the `@dlt.source` in `ol_dlt/sources/<name>/__init__.py` — no env
    branching, no Dagster imports. Resolve destination/dataset/table_format via
-   `ol_dlt.config`.
+   `ol_dlt.config`. For a relational source, declare a `DatabaseSourceSpec`
+   instead (see **Database sources** above).
 2. Add a `@dlt_assets` wrapper in the `data_loading` code location
    (`defs/ingestion/assets.py`) using the shared `RawDataDltTranslator`.

src/ol_dlt/ol_dlt/config.py

@@ -65,14 +65,19 @@ def active_profile() -> str:
     return os.getenv("DLT_PROFILE", DEFAULT_PROFILE)
 
 
-def active_table_format() -> TableFormat:
-    """Return the table format for the active profile.
+def active_table_format(profile: str | None = None) -> TableFormat:
+    """Return the table format for the (active) profile.
 
     ``"iceberg"`` for qa/production, ``"native"`` (plain filesystem parquet) for
     dev/ci/test. Use this in ``@dlt.resource(..., table_format=...)`` so a source
     body never branches on the environment itself.
+
+    Takes an optional explicit ``profile`` for the same reason ``dataset_name``
+    and ``bucket_root`` do: a caller that has already resolved a profile (a test,
+    or a targeted run) must not have the table format silently resolved from a
+    different one via ``DLT_PROFILE``.
     """
-    return "iceberg" if active_profile() in ICEBERG_PROFILES else "native"
+    return "iceberg" if (profile or active_profile()) in ICEBERG_PROFILES else "native"
 
 
 def dataset_name(profile: str | None = None) -> str: