Add dlt-based ingestion app (app_etl_dlt)

apps/app_etl_dlt/README.md

@@ -0,0 +1,133 @@
+# app_etl_dlt — dlt-based ingestion jobs
+
+Same goal as `app_etl` (Postgres read replica → GCS Parquet → BigQuery,
+cheaply and correctly), letting dlt own the boilerplate `app_etl` hand-rolled:
+chunked cursor reads, streaming Parquet writes, BigQuery batch load jobs, and
+incremental-cursor state tracking.
+
+This is a fresh app, not a rewrite of `app_etl` in place — both exist so the
+approaches can be compared before deciding which one stays.
+
+## Design: config registry + generic runner
+
+One table = one `TableConfig` entry in `tables.py`. `runner.py` has exactly
+two execution paths — `run_incremental` and `run_snapshot` — and dispatches
+to whichever the table's `mode` says. Adding a table means adding a
+`TableConfig`, not a new job file:
+
+```python
+# tables.py
+TABLES: list[TableConfig] = [
+    TableConfig(
+        name="payments",
+        mode="incremental",
+        cadence="hourly",
+        cursor_column="created_at",
+        initial_value=datetime(2025, 1, 1, tzinfo=timezone.utc),
+        primary_key="id",
+    ),
+    TableConfig(name="merchants", mode="snapshot", cadence="daily"),
+]
+```
+
+Run a table:
+
+```bash
+uv run python -m app_etl_dlt.runner payments
+```
+
+`TABLES` ships **empty** — add whichever tables you're actually ingesting;
+there's no fixed pair this app expects.
+
+`cadence` is documentation only for now: it says how the Cloud Scheduler job
+invoking `runner.py <name>` should be set up. Nothing in code reads it —
+there's no scheduler wired up yet.
+
+## What dlt replaces from `app_etl`
+
+| `app_etl` hand-rolled | dlt equivalent |
+|---|---|
+| `utils/pg.py` — named-cursor chunked reads into pyarrow batches | `sql_table(..., chunk_size=...)` |
+| `utils/gcs.py` — streaming Parquet writer to a fixed GCS layout | `staging=dlt.destinations.filesystem(bucket_url=...)` |
+| `utils/bq.py` — `WRITE_APPEND` / `WRITE_TRUNCATE` load jobs | `pipeline.run(..., write_disposition=...)` (still a free batch load job, not streaming inserts) |
+| `utils/state.py` + `ops.ingestion_runs` watermark | dlt's own pipeline state, versioned in the destination, advanced only after a successful load |
+| one job file per table | one `TableConfig` entry per table |
+
+The incremental-window correctness contract carries over unchanged: Postgres
+`now()` is transaction *start* time, so a cursor column alone isn't a safe
+watermark. `run_incremental` in `runner.py` keeps the same guard `app_etl`
+uses — the window's upper bound trails wall clock by `safety_lag_minutes`
+(default 10) instead of extracting up to `now()`. See `app_etl`'s README for
+the full race-condition writeup; the reasoning is identical here.
+
+## jsonb/json columns
+
+BigQuery can't load dlt's `json` data type from Parquet files (only from
+`jsonl`/`model`) — confirmed against dlt 1.28's `ensure_supported_type`,
+which raises rather than silently mis-loading or dropping the column. Any
+table with `jsonb`/`json` Postgres columns (e.g. `payments.risk`,
+`payments.customer`) needs `has_json_columns=True` set on its `TableConfig`
+— `runner.py` uses that to apply BigQuery's `autodetect_schema` hint, which
+lets BigQuery infer column types straight from the Parquet file instead of
+dlt pre-declaring them. Those columns land as BigQuery `STRING` (raw JSON
+text, queryable via `JSON_EXTRACT`/`PARSE_JSON`), not a native `JSON` column.
+Forgetting this flag fails the whole load, loudly, not silently — so it's
+hard to miss when adding a new table.
+
+## Known gaps vs. `app_etl` (deliberate, not accidental)
+
+- **No per-day snapshot history.** `app_etl`'s `ingest_merchants` loads into
+  `merchants$YYYYMMDD` (a BigQuery partition decorator) with `WRITE_TRUNCATE`,
+  so every day's snapshot survives for point-in-time joins. `run_snapshot`'s
+  `write_disposition="replace"` only replaces the whole table each run —
+  idempotent per run, but no history. Closing this gap means a
+  `bigquery_adapter` partition hint plus a partition-scoped load call, which
+  re-introduces the hand-rolled BigQuery API code this app exists to avoid.
+  Not implemented; revisit per-table if point-in-time history turns out to
+  matter for a given snapshot table.
+- **No queryable run log.** `app_etl` writes an explicit, append-only
+  `ops.ingestion_runs` row per run (`rows_loaded`, `gcs_uri`, `watermark`) for
+  recon and debugging before an orchestrator exists. dlt's equivalent is the
+  `LoadInfo` object `run_table()` returns (printed by the CLI, not persisted).
+  If that observability is needed, write it into a BigQuery table shaped like
+  `ops.ingestion_runs` — not implemented here.
+
+## Configuration
+
+`config.py` reads env vars: `GCP_PROJECT`, `GCS_BUCKET`, `BQ_DATASET_RAW`
+(default `raw_litecore`), `BQ_LOCATION` (default `me-central2` — **must
+match the GCS bucket's region**; BigQuery load jobs require the dataset and
+the source bucket to be co-located, otherwise you'll hit a confusing
+"dataset not found in location US" error even though the dataset exists),
+and Postgres as four separate parts — `PG_HOST` (default `127.0.0.1`),
+`PG_PORT` (default `5432`), `PG_USER`, `PG_DATABASE` — rather than one DSN
+string, so switching databases (this instance hosts one per service) is a
+one-line env change. `Settings.pg_dsn` assembles the SQLAlchemy DSN from
+these at call time.
+
+No password field: this assumes Cloud SQL IAM database auth through a
+locally running Cloud SQL Auth Proxy —
+
+```bash
+cloud-sql-proxy --auto-iam-authn --port=5432 <project>:<region>:<instance>
+```
+
+`PG_USER` is your IAM identity (e.g. your `@lite.sa` email) — the proxy
+handles the actual auth handshake with Cloud SQL, so nothing here ever
+holds a static DB password. See `.env` for a filled-in example.
+
+## Status
+
+Working — `run_incremental` has been run end-to-end against real
+infrastructure (`payment_v2.payments` in Cloud SQL → `lite-data-dev-raw` GCS
+staging → `lite-data-dev.raw_litecore.payments` in BigQuery, `me-central2`),
+including the `has_json_columns` path. `run_snapshot` is implemented the same
+way but hasn't had a real run yet. `TABLES` currently has one entry
+(`payments`) — add more as needed.
+
+Required IAM on the target GCP project, learned the hard way: dataset
+creation alone isn't enough — you need `roles/bigquery.dataEditor` (tables,
+not just datasets), `roles/bigquery.jobUser` (to actually run load jobs;
+`dataEditor` doesn't include this), and `roles/storage.objectAdmin` (or at
+least `objectCreator`) on the staging bucket. Missing any one of these fails
+at a different, confusingly-specific step rather than up front.

apps/app_etl_dlt/app_etl_dlt/config.py

@@ -0,0 +1,60 @@
+"""Environment-driven settings — one flat object, no layering.
+
+Same env-var contract as app_etl (single dev GCP project for now). dlt
+gets what it needs (GCS staging bucket, BigQuery dataset, PG connection)
+translated from these at each pipeline's call site — no dlt.toml, no
+dlt-managed secrets file, one source of truth for config across both
+apps.
+
+Postgres connection is split into PG_HOST/PORT/USER/DATABASE rather than
+one PG_DSN string, so switching databases (this instance hosts one per
+service: payment_v2, risk_management, ledger, ...) is a one-line env
+change instead of rebuilding a DSN. No password field: this assumes
+Cloud SQL IAM database auth through a locally running Cloud SQL Auth
+Proxy (`cloud-sql-proxy --auto-iam-authn`) — the same passwordless
+mechanism LiteCore's own services use in cloud. PG_USER is your IAM
+identity (e.g. your @lite.sa email); the proxy handles the actual auth
+handshake, so nothing here ever holds a DB password.
+"""
+
+from __future__ import annotations
+
+import os
+import urllib.parse
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Settings:
+    gcp_project: str      # GCP project id (single dev project for now)
+    gcs_bucket: str        # dlt filesystem staging root: "bucket" or "bucket/prefix"
+                           # if you don't have create-bucket access, point this at a
+                           # folder in an existing bucket, e.g. "old-bucket/lite-data-platform"
+    bq_dataset_raw: str    # landing dataset in BQ, e.g. raw_litecore
+    bq_location: str        # must match the GCS bucket's region — BigQuery load jobs
+                            # require the dataset and source bucket to be co-located
+    pg_host: str           # e.g. 127.0.0.1 (Cloud SQL Auth Proxy, run separately)
+    pg_port: int           # e.g. 5432
+    pg_user: str           # IAM identity authorized on the instance, e.g. you@lite.sa
+    pg_database: str       # e.g. payment_v2, risk_management — the one thing you'll change often
+
+    @property
+    def pg_dsn(self) -> str:
+        """SQLAlchemy-style DSN for dlt's sql_table(credentials=...). Builds
+        it from the parts above so callers never hand-encode the '@' in an
+        IAM email themselves."""
+        user = urllib.parse.quote(self.pg_user, safe="")
+        return f"postgresql+psycopg://{user}@{self.pg_host}:{self.pg_port}/{self.pg_database}"
+
+    @classmethod
+    def from_env(cls) -> Settings:
+        return cls(
+            gcp_project=os.environ["GCP_PROJECT"],
+            gcs_bucket=os.environ["GCS_BUCKET"],
+            bq_dataset_raw=os.environ.get("BQ_DATASET_RAW", "raw_litecore"),
+            bq_location=os.environ.get("BQ_LOCATION", "me-central2"),
+            pg_host=os.environ.get("PG_HOST", "127.0.0.1"),
+            pg_port=int(os.environ.get("PG_PORT", "5432")),
+            pg_user=os.environ["PG_USER"],
+            pg_database=os.environ["PG_DATABASE"],
+        )

apps/app_etl_dlt/app_etl_dlt/runner.py

@@ -0,0 +1,111 @@
+"""Generic ingestion runner — dispatches a TableConfig (see tables.py) to
+one of two dlt pipeline shapes, incremental or snapshot. Adding a table
+should never require a branch in this file; it means a new TableConfig.
+
+Incremental correctness contract (unchanged from app_etl's hand-rolled
+version — see that app's README for the full writeup): Postgres now() is
+transaction *start* time, so a cursor column alone is not a safe
+watermark — a row can commit after a later run has already advanced past
+its cursor value. The guard: never extract up to now(); trail the
+window's upper bound by `safety_lag_minutes` instead. dlt's own
+incremental state (versioned in the destination) tracks the watermark
+between runs — no hand-rolled ops.ingestion_runs table.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+
+import dlt
+from dlt.common.pipeline import LoadInfo
+from dlt.destinations.adapters import bigquery_adapter
+from dlt.extract.resource import DltResource
+from dlt.sources.sql_database import sql_table
+
+from app_etl_dlt.config import Settings
+from app_etl_dlt.tables import TABLES, TableConfig
+
+CHUNK_ROWS = 50_000
+
+
+def _pipeline(table_name: str, settings: Settings) -> dlt.Pipeline:
+    return dlt.pipeline(
+        pipeline_name=f"ingest_{table_name}",
+        destination=dlt.destinations.bigquery(
+            project_id=settings.gcp_project, location=settings.bq_location
+        ),
+        staging=dlt.destinations.filesystem(bucket_url=f"gs://{settings.gcs_bucket}"),
+        dataset_name=settings.bq_dataset_raw,
+    )
+
+
+def _apply_json_hint(resource: DltResource, table: TableConfig) -> DltResource:
+    """BigQuery can't load dlt's "json" data_type from Parquet files (only
+    from jsonl/model) — verified against dlt 1.28's ensure_supported_type,
+    which raises rather than silently mis-loading. autodetect_schema lets
+    BigQuery infer column types from the Parquet file itself instead of dlt
+    pre-declaring them, which is dlt's own documented way out. Only applied
+    when the table actually has json/jsonb columns, so every other table
+    keeps dlt's normal (more precise) type inference."""
+    if table.has_json_columns:
+        return bigquery_adapter(resource, autodetect_schema=True)
+    return resource
+
+
+def run_incremental(table: TableConfig, settings: Settings) -> LoadInfo:
+    end_value = datetime.now(timezone.utc) - timedelta(minutes=table.safety_lag_minutes)
+    resource = sql_table(
+        credentials=settings.pg_dsn,
+        table=table.name,
+        incremental=dlt.sources.incremental(
+            table.cursor_column,
+            initial_value=table.initial_value,
+            end_value=end_value,
+        ),
+        chunk_size=CHUNK_ROWS,
+    )
+    resource = _apply_json_hint(resource, table)
+    return _pipeline(table.name, settings).run(
+        resource,
+        table_name=table.name,
+        write_disposition="append",
+        primary_key=table.primary_key,
+        loader_file_format="parquet",
+    )
+
+
+def run_snapshot(table: TableConfig, settings: Settings) -> LoadInfo:
+    resource = sql_table(credentials=settings.pg_dsn, table=table.name, chunk_size=CHUNK_ROWS)
+    resource = _apply_json_hint(resource, table)
+    return _pipeline(table.name, settings).run(
+        resource,
+        table_name=table.name,
+        write_disposition="replace",
+        loader_file_format="parquet",
+    )
+
+
+def run_table(table_name: str, settings: Settings | None = None) -> LoadInfo:
+    table = next((t for t in TABLES if t.name == table_name), None)
+    if table is None:
+        known = ", ".join(t.name for t in TABLES) or "(none configured)"
+        raise ValueError(f"unknown table '{table_name}'; configured tables: {known}")
+
+    settings = settings or Settings.from_env()
+    if table.mode == "incremental":
+        return run_incremental(table, settings)
+    return run_snapshot(table, settings)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run one table's ingestion job.")
+    parser.add_argument("table", help="table name as configured in tables.py")
+    args = parser.parse_args()
+
+    load_info = run_table(args.table)
+    print(load_info)
+
+
+if __name__ == "__main__":
+    main()

apps/app_etl_dlt/app_etl_dlt/tables.py

@@ -0,0 +1,62 @@
+"""Table registry — the one place per-table ingestion is configured.
+
+Add a table by appending one `TableConfig` here; `runner.py`'s two generic
+paths (incremental / snapshot) do the rest. dlt already absorbs the
+per-table plumbing (cursor chunking, parquet staging, load jobs), so the
+only thing that varies table-to-table is this config — no new job file
+per table.
+
+`cadence` is documentation only right now: it says how the Cloud Scheduler
+job invoking `runner.py <name>` should be set up. Nothing in code reads it
+yet — there's no scheduler wired up.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Literal
+
+
+@dataclass(frozen=True)
+class TableConfig:
+    name: str  # Postgres table name; also the BigQuery destination table name
+    mode: Literal["incremental", "snapshot"]
+    cadence: str  # informational only, e.g. "hourly", "daily", "*/5 * * * *"
+
+    # incremental-only fields
+    cursor_column: str | None = None  # e.g. "created_at" or "updated_at"
+    initial_value: datetime | None = None  # launch date / earliest row to backfill from
+    primary_key: str | None = None
+    safety_lag_minutes: int = 10  # see runner.py's module docstring for why this exists
+
+    # Set True if the table has jsonb/json columns. BigQuery can't load
+    # dlt's "json" data_type from Parquet files (only from jsonl/model) —
+    # this tells runner.py to apply BigQuery's autodetect_schema hint,
+    # which lets BigQuery infer types from the Parquet file directly instead
+    # of dlt pre-declaring them. Verified against dlt 1.28's actual source
+    # (dlt/destinations/impl/bigquery/factory.py's ensure_supported_type):
+    # without it, loading fails outright, not silently.
+    has_json_columns: bool = False
+
+    def __post_init__(self) -> None:
+        if self.mode == "incremental" and (self.cursor_column is None or self.initial_value is None):
+            raise ValueError(f"{self.name}: incremental tables need cursor_column and initial_value")
+
+
+# One entry per ingested table. Empty by default — add your own.
+TABLES: list[TableConfig] = [
+    TableConfig(
+        name="payments",
+        mode="incremental",
+        cadence="hourly",
+        cursor_column="created_at",
+        # earliest row currently in payment_v2.payments is 2026-05-06;
+        # start a little before that so a first run backfills everything.
+        initial_value=datetime(2026, 5, 1, tzinfo=timezone.utc),
+        primary_key="id",
+        # risk, customer, order_data, device, threeds_input, threeds_result,
+        # return_url, metadata, routing_result, risk_result are all jsonb.
+        has_json_columns=True,
+    ),
+]

apps/app_etl_dlt/pyproject.toml

@@ -0,0 +1,24 @@
+[project]
+name = "app-etl-dlt"
+version = "0.1.0"
+description = "dlt-based ingestion jobs: Postgres read replica -> GCS (Parquet staging) -> BigQuery."
+requires-python = ">=3.14"
+# Extras verified against dlt 1.28.0's PyPI metadata: "sql-database" (hyphen,
+# not underscore) for the sql_database source, "gs" for GCS filesystem
+# staging ("filesystem" alone pulls s3fs/botocore, not gcsfs). Deliberately
+# omitting the "postgres" extra — it pulls psycopg2-binary, which would sit
+# alongside the psycopg (v3) driver app_etl already uses; psycopg[binary]
+# below covers the driver instead, via the "postgresql+psycopg://" DSN
+# scheme (SQLAlchemy 2.0's native psycopg3 dialect).
+dependencies = [
+    "dlt[bigquery,gs,sql-database]>=1.4",
+    "sqlalchemy>=2",
+    "psycopg[binary]>=3.2",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["app_etl_dlt"]