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

[blarghmatey]

What are the relevant tickets?

N/A

Description (What does it do?)

Adds a reusable relational-database source to ol_dlt, and uses the Keycloak Postgres database as its first instance. Keycloak is a good first case because it is net-new — it has no Airbyte connection and therefore no existing state or cursor to seed, so nothing has to be cut over.

The reusable part — src/ol_dlt/ol_dlt/database.py

A source is a declaration, not a pipeline. A DatabaseSourceSpec of DatabaseTable entries becomes a @dlt.source built on dlt's sql_table, with resources renamed to the warehouse's raw-table convention (raw__<system>__app__postgres__<table>) so the existing RawDataDltTranslator produces correct asset keys and groups with no changes:

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"), ...),
)

Credentials — src/ol_dlt/ol_dlt/vault.py

Two constraints shaped this, and they are the part most worth reviewing:

1. Dagster instantiates every dlt source when the code location is imported (build_source() runs at module scope in assets.py). A Vault lease taken there would be long stale — or revoked — by the time a scheduled run actually fires. So credentials are injected per connection, via a SQLAlchemy do_connect event listener registered through sql_table(engine_adapter_callback=...).
2. For the same reason, table reflection is deferred (defer_table_reflect=True) so no database connection is opened at import either. Resource names are set explicitly with .with_name(), so Dagster asset keys remain statically known despite the deferral.

The connection URL for deployed profiles carries only a host and database name — a username or password never appears in it — and forces sslmode=require (the RDS parameter group sets rds.force_ssl=1). Credentials come from <mount>/creds/readonly and are cached for half the lease, so one lease serves every table in a run rather than one lease per table.

vault.py is pure hvac and mirrors the environment contract of authenticate_vault (DAGSTER_VAULT_ROLE / DAGSTER_VAULT_MOUNT, Kubernetes auth in-cluster and VAULT_TOKEN outside it), because ol_dlt must not import Dagster — enforced by the existing ruff TID251 banned-api rule.

Environment alignment

Which database gets read follows the deployment, not the spec. DLT_PROFILE (derived from DAGSTER_ENVIRONMENT) selects it: qa reads the QA database, production reads production, and dev/ci/test read the local-dev CloudNativePG cluster that ol-infrastructure/local-dev stands up. A developer never needs access to a deployed environment to exercise a pipeline. Per-source overrides come from <NAME>_DB_HOST / _PORT / _USERNAME / _PASSWORD.

Keycloak scope

A curated identity subset (12 tables: users, attributes, group membership, role assignment, realm/client context) rather than the whole schema. Keycloak's remaining ~80 tables are its own internal bookkeeping, and several hold credential material. Deliberately excluded, with a test asserting the exclusions hold:

Excluded	Why
credential	password hashes and OTP secrets
component_config	identity-provider and LDAP bind secrets
*_session, *_event	high-churn operational state, not identity facts
client.secret, client.registration_token	column-level exclusion; client credentials

Every table uses write_disposition="replace". Keycloak records no reliable row-level modification time — user_entity.created_timestamp is creation-only, while emails, names and enablement all change in place — so reconciling missed updates would cost more than a full re-read of a small realm. The spec supports incremental cursors (DatabaseTable.cursor_column, which requires a primary key to merge on) for future sources that do have one.

Scheduled daily at 04:30 UTC, selecting by asset group rather than by key so adding a table to the spec does not also require editing the schedule.

How can this be tested?

Automated (from src/ol_dlt, no credentials or network needed):

cd src/ol_dlt && uv sync && uv run pytest

69 tests pass. The ones that matter most for this change:

• test_database.py::test_materializes_a_database_to_the_test_profile — end-to-end reflect → extract → write against a real relational source. SQLite stands in for Postgres so it stays hermetic; everything under test (deferred reflection, resource renaming, column exclusion, raw table layout) is driver-independent. It asserts an excluded column is absent from the written schema.
• test_database.py::test_deployed_profile_never_reads_vault_at_definition_time — replaces read_database_credentials with a function that fails the test if called, then builds the production source. This is the regression guard for constraint (1) above.
• test_database.py::test_vault_credentials_are_injected_at_connection_time — fires the do_connect event and asserts the connect params carry the Vault-issued user and password.
• test_database.py::test_deployed_profile_url_omits_credentials_and_requires_tls — asserts the URL has no username or password and does set sslmode=require.
• test_vault.py — address derivation (dev points at QA Vault, never production), lease caching and expiry, and that a denied path names the policy file to fix.
• test_keycloak.py — the forbidden-table and secret-column exclusions, and the raw naming convention.

Also verified: uv run ruff check --select TID251 . passes (no Dagster imported), and uv tree --package ol-dlt | grep -i dagster is empty — the package/orchestration separation still holds.

Dagster code location loads and resolves the assets:

cd dg_projects/data_loading && uv sync
DAGSTER_ENVIRONMENT=dev uv run python -c "
from data_loading.definitions import defs
from dagster import AssetSelection
g = defs.resolve_asset_graph()
print(len(AssetSelection.groups('keycloak').resolve(g)), 'keycloak assets')
print(defs.resolve_schedule_def('keycloak_ingest_daily_schedule').cron_schedule)
"

Expect 12 keycloak assets and 30 4 * * *.

Live run against local-dev (needs the ol-infrastructure local-dev stack up):

kubectl port-forward -n local-infra svc/local-pg-rw 5432:5432
cd src/ol_dlt && DLT_PROFILE=dev uv run python -m ol_dlt.sources.keycloak
uv run dlt pipeline keycloak info

Note that a freshly bootstrapped local-dev keycloak database is empty until Keycloak has started against it and created its schema; deferred reflection means the failure in that case is a clear "no such table" rather than something subtle.

pre-commit passes on all changed files (ruff format, ruff check, mypy, detect-secrets).

Additional Context

Requires a companion infrastructure PR before deployed runs work: mitodl/ol-infrastructure#5092 — it grants postgres-keycloak/creds/readonly in the Dagster Vault policy and passes KEYCLOAK_DB_HOST through from the Keycloak stack export. Until that merges and deploys, a QA or production run of this pipeline fails immediately with a message naming exactly what is missing (no partial writes). The dev profile is unaffected and works today. Each future database source needs the same two additions there, and the README documents that.

Two things intentionally left out of scope, worth deciding before the first production run:

1. No dbt sources yet. There is no ol_warehouse_raw_data source YAML or staging model for these tables. That is a separate change once the raw tables actually exist and their shape is known.
2. PII. user_entity and user_attribute carry email addresses and names into the raw layer. This PR applies no masking or access restriction beyond keeping credential material out entirely. If the raw layer needs column-level handling for those, it should be settled before this runs against production rather than retrofitted after data has landed.

On CDC, which the migration notes flag as the eventual question: this source is cursor/replace-based and read-replica-safe by design. It captures no deletes. For Keycloak that is unlikely to matter much (user deletion is rare and enabled is the operative flag), but it is a real gap for any future source where deletes are meaningful, and periodic reconciliation anti-joins are the cheaper first answer there.

[gitguardian]

⚠️ GitGuardian has uncovered 1 secret following the scan of your pull request.

Please consider investigating the findings and remediating the incidents. Failure to do so may lead to compromising the associated services or software components.

🔎 Detected hardcoded secret in your pull request

GitGuardian id	GitGuardian status	Secret	Commit	Filename
35074009	Triggered	Generic Password	f60407a	src/ol_dlt/tests/test_vault.py	View secret

🛠 Guidelines to remediate hardcoded secrets

1. Understand the implications of revoking this secret by investigating where it is used in your code.
2. Replace and store your secret safely. Learn here the best practices.
3. Revoke and rotate this secret.
4. If possible, rewrite git history. Rewriting git history is not a trivial act. You might completely break other contributing developers' workflow and you risk accidentally deleting legitimate data.

To avoid such incidents in the future consider

• following these best practices for managing and storing secrets including API keys and other credentials
• install secret detection on pre-commit to catch secret before it leaves your machine and ease remediation.

---

^{🦉 GitGuardian detects secrets in your source code to help developers and security teams secure the modern development process. You are seeing this because you or someone else with access to this repository has authorized GitGuardian to scan your pull request.}

[Copilot]

Pull request overview

Adds a reusable relational-database ingestion abstraction to ol_dlt, and introduces Keycloak Postgres as the first concrete database-backed source, including Vault-issued credential handling and Dagster scheduling.

Changes:

• Added ol_dlt.database to build @dlt.source objects from a declarative DatabaseSourceSpec (with raw-table naming and optional incremental configuration).
• Added ol_dlt.vault (pure hvac) to fetch and cache short-lived DB credentials and inject them at SQLAlchemy connection time.
• Added Keycloak database source + Dagster asset wrapper and a daily schedule, with tests asserting table/column exclusions and deferred Vault access.

Reviewed changes

Copilot reviewed 11 out of 13 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/ol_dlt/ol_dlt/database.py	New reusable database source builder using dlt sql_table, deferred reflection, and per-connection credential injection.
src/ol_dlt/ol_dlt/vault.py	New Vault credential resolver (hvac) with lease-aware caching and clear error messaging for missing/denied paths.
src/ol_dlt/ol_dlt/sources/keycloak/__init__.py	Declares KEYCLOAK_SPEC and exposes a uniform build_source() entrypoint for Dagster.
src/ol_dlt/ol_dlt/sources/keycloak/__main__.py	Adds a standalone smoke-run entrypoint for local-dev execution.
src/ol_dlt/tests/test_database.py	Adds unit/integration coverage for spec validation, URL construction, Vault injection timing, and column exclusions.
src/ol_dlt/tests/test_vault.py	Adds tests for Vault address derivation, caching behavior, and error handling.
src/ol_dlt/tests/sources/test_keycloak.py	Adds regression tests for forbidden-table and secret-column exclusions plus raw naming.
src/ol_dlt/pyproject.toml	Adds dependencies/extras needed for SQL database ingestion and Vault (dlt sql_database, psycopg2-binary, hvac).
src/ol_dlt/uv.lock	Locks new dependencies/extras for ol_dlt.
src/ol_dlt/README.md	Documents how to declare and run database sources and what infra changes are required for deployments.
dg_projects/data_loading/data_loading/defs/ingestion/assets.py	Registers Keycloak ingestion assets in the Dagster data_loading code location.
dg_projects/data_loading/data_loading/defs/ingestion/schedules.py	Adds a daily schedule targeting the keycloak asset group.
dg_projects/data_loading/uv.lock	Updates the code location lockfile to include the updated ol_dlt dependency set.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.