diff --git a/.github/workflows/docker-smoke-test.yml b/.github/workflows/docker-smoke-test.yml index d688a800..ca35e2e9 100644 --- a/.github/workflows/docker-smoke-test.yml +++ b/.github/workflows/docker-smoke-test.yml @@ -28,7 +28,7 @@ jobs: echo "Building $dockerfile in directory $dir..." context="$dir" case "$dockerfile" in - ./images/dagster/*|./images/superset/*|./images/dbt/*) context="." ;; + ./images/dagster/*|./images/superset/*|./images/dbt/*|./images/dlt/*) context="." ;; esac docker build -f "$dockerfile" "$context" || exit 1 done diff --git a/.github/workflows/image-security-scan.yml b/.github/workflows/image-security-scan.yml index 99804fb3..7bc33372 100644 --- a/.github/workflows/image-security-scan.yml +++ b/.github/workflows/image-security-scan.yml @@ -82,7 +82,7 @@ jobs: dockerfile="${{ matrix.image.dockerfile }}" context="$(dirname "$dockerfile")" case "$dockerfile" in - images/dagster/*|images/superset/*|images/dbt/*) context="." ;; + images/dagster/*|images/superset/*|images/dbt/*|images/dlt/*) context="." ;; esac tag="${{ matrix.image.name }}${{ matrix.image.variant != '' && format('-{0}', matrix.image.variant) || '' }}" docker build -f "$dockerfile" -t "cds/$tag:scan" "$context" diff --git a/.github/workflows/publish-images.yml b/.github/workflows/publish-images.yml index 0ea73695..098160e0 100644 --- a/.github/workflows/publish-images.yml +++ b/.github/workflows/publish-images.yml @@ -98,7 +98,7 @@ jobs: dockerfile="${{ matrix.image.dockerfile }}" context="$(dirname "$dockerfile")" case "$dockerfile" in - images/dagster/*|images/superset/*|images/dbt/*) context="." ;; + images/dagster/*|images/superset/*|images/dbt/*|images/dlt/*) context="." ;; esac variant="${{ matrix.image.variant }}" prefix="" @@ -258,6 +258,9 @@ jobs: dbt) base_version="$(grep -m1 '^dbt-core==' images/dbt/requirements.txt | cut -d= -f3)" ;; + dlt) + base_version="$(grep -m1 '^dlt\[postgres\]==' images/dlt/requirements.txt | sed -E 's/^dlt\[postgres\]==([^[:space:]]+).*/\1/')" + ;; *) echo "No version scheme defined for image '${{ matrix.image.name }}'" >&2 exit 1 @@ -300,7 +303,7 @@ jobs: dockerfile="${{ matrix.image.dockerfile }}" context="$(dirname "$dockerfile")" case "$dockerfile" in - images/dagster/*|images/superset/*|images/dbt/*) context="." ;; + images/dagster/*|images/superset/*|images/dbt/*|images/dlt/*) context="." ;; esac ref="docker.io/ronaldsoeverein/${{ matrix.image.name }}" tag="${{ steps.version.outputs.tag }}" diff --git a/docs/architecture.md b/docs/architecture.md index 593dec68..0f96828b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -39,6 +39,7 @@ The platform is divided into logical layers. | --- | --- | --- | | **Secrets** | credentials, secret injection, key management | Vault | | **Infrastructure services** | service coordination backends and support systems | KeyDB | +| **Ingestion** | extract-load pipelines into the warehouse | dlt (experimental) | | **Storage / compute** | databases, warehouses, and processing engines | Postgres, DuckDB (experimental) | | **Orchestration** | workflow scheduling and task execution | Dagster | | **Transformation** | data modeling and transformation | dbt (experimental; targets Postgres or DuckDB) | diff --git a/docs/image-scanning.md b/docs/image-scanning.md index aa8a6a47..505824ad 100644 --- a/docs/image-scanning.md +++ b/docs/image-scanning.md @@ -31,7 +31,8 @@ Release object for them: - a base version derived from the pinned upstream dependency (`dagster==` in `images/dagster/requirements.txt`, `FROM apache/superset:` in `images/superset/base/Dockerfile`, `dbt-core==` in - `images/dbt/requirements.txt`), + `images/dbt/requirements.txt`, `dlt[postgres]==` in + `images/dlt/requirements.txt`), - an optional `-` prefix for non-default image variants, - plus a `sha-<12-char-commit-sha>` tag (immutable, always pushed) and a `latest`/`latest` tag. diff --git a/docs/image-signing.md b/docs/image-signing.md index 5085368e..5a20b4ed 100644 --- a/docs/image-signing.md +++ b/docs/image-signing.md @@ -19,8 +19,8 @@ differ (see below). ### Docker Hub - Registry: `docker.io` (`registry-1.docker.io`) -- Name: `docker.io/ronaldsoeverein/` (e.g. `dagster`, `superset`, `dbt`; no `cds-` prefix). -- Tags: a base-version tag derived per image (`dagster==`/`apache/superset:`/`dbt-core==` version) and `latest`, both optionally prefixed with the build variant (e.g. `hardened-1.8.0`, `hardened-latest`). The `hardened` variant additionally gets an `alpine-` aliased pair (`alpine-1.8.0`, `alpine-latest`) pointing at the same image. As with GHCR, tags are mutable; verification is always by digest. +- Name: `docker.io/ronaldsoeverein/` (e.g. `dagster`, `superset`, `dbt`, `dlt`; no `cds-` prefix). +- Tags: a base-version tag derived per image (`dagster==`/`apache/superset:`/`dbt-core==`/`dlt[postgres]==` version) and `latest`, both optionally prefixed with the build variant (e.g. `hardened-1.8.0`, `hardened-latest`). The `hardened` variant additionally gets an `alpine-` aliased pair (`alpine-1.8.0`, `alpine-latest`) pointing at the same image. As with GHCR, tags are mutable; verification is always by digest. ## Verifying an image diff --git a/docs/roadmap.md b/docs/roadmap.md index 55dea758..30cc6298 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -27,6 +27,7 @@ These are considered production-ready in the current release (v0.4.0): These work but may have breaking changes in upcoming releases: - Module: Airflow (`modules-experimental/orchestration/airflow/`) — not yet integrated into a stable profile +- Module: dlt (`modules-experimental/ingestion/dlt/`) — one-shot pipeline job, not yet wired into a stable profile - Module: DuckDB (`modules-experimental/warehouse/duckdb/`) — embedded/file-based warehouse via the new `file-database` contract; wired into dbt (#599), not yet wired into dlt or a demo profile (#593) - Module: dbt (`modules-experimental/transformation/dbt/`) — one-shot dbt-core transformation job; targets Postgres or DuckDB via `config.warehouseType` (#599) - `cds test` — implemented; not yet exercised in CI or real contributor usage diff --git a/images/dlt/Dockerfile b/images/dlt/Dockerfile new file mode 100644 index 00000000..c83750f2 --- /dev/null +++ b/images/dlt/Dockerfile @@ -0,0 +1,47 @@ +FROM python:3.14-slim@sha256:cad9a2c871761c413caa6fdd6441c783451e740a48aaeba60ae62a8b53525ef6 AS builder + +ENV PIP_DISABLE_PIP_VERSION_CHECK=1 \ + PYTHONUNBUFFERED=1 +WORKDIR /app + +RUN python -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + +COPY images/dlt/requirements.txt ./ +# Uninstall pip once dependencies are installed; pip's bundled CycloneDX SBOM +# lists vendored build dependencies (e.g. msgpack, setuptools) at versions +# Trivy flags as vulnerable even though dlt never imports them. Matches +# images/dagster/base/Dockerfile's and images/dbt/Dockerfile's pattern. +RUN python -m pip install --no-cache-dir -r requirements.txt && \ + python -m pip uninstall --yes pip + +FROM python:3.14-slim@sha256:cad9a2c871761c413caa6fdd6441c783451e740a48aaeba60ae62a8b53525ef6 + +ENV HOME=/home/dlt \ + PATH="/opt/venv/bin:$PATH" \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +# Pull in trixie-security fixes not yet baked into the pinned +# python:3.14-slim digest, matching images/superset/base/Dockerfile's and +# images/dbt/Dockerfile's apt-get upgrade pattern. +RUN apt-get update \ + && apt-get upgrade -y --no-install-recommends \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/* + +RUN groupadd --system --gid 999 dlt && \ + useradd --system --uid 999 --no-create-home --gid dlt dlt && \ + install -d -o dlt -g dlt "$HOME" "$HOME/.dlt" /usr/app/dlt_state && \ + python -m pip uninstall --yes pip 2>/dev/null || true + +WORKDIR /app + +COPY --from=builder /opt/venv /opt/venv +COPY --chmod=0555 images/dlt/entrypoint.sh /entrypoint.sh + +USER dlt + +ENTRYPOINT ["/entrypoint.sh"] +# No CMD — DLT_ENTRYPOINT (set via the module's compose environment) +# selects the pipeline script to run; see entrypoint.sh. diff --git a/images/dlt/README.md b/images/dlt/README.md new file mode 100644 index 00000000..05042084 --- /dev/null +++ b/images/dlt/README.md @@ -0,0 +1,47 @@ +# CDS dlt Image + +Custom image for the [Composable Data Stack (CDS)](https://github.com/RonaldHensbergen/composable-data-stack) +`modules-experimental/ingestion/dlt` module. Runs a bind-mounted Python +pipeline script with [dlt (data load tool)](https://dlthub.com/) as a +one-shot job, loading into a consumed `sql-database` contract (e.g. the +`postgres` module). + +> This image is built and wired automatically by `cds render`/`cds up` for +> profiles that include the `dlt` module — you normally never invoke it +> directly. It is experimental, built locally via the `build:` block in +> `module.yaml`, and (like `images/dagster`/`images/superset`/`images/dbt`) +> also built, scanned, signed, and published to a registry by the repo's +> image workflows. + +## Configuration + +### Environment variables + +| Variable | Required | Purpose | +| --- | --- | --- | +| `DESTINATION__POSTGRES__CREDENTIALS` | yes | Full `postgresql://` connection URI, sourced from the module's consumed `sql-database` contract binding. dlt's postgres destination reads this exact variable name itself. | +| `DLT_PROJECT_DIR` | baked in | Pipeline project directory (default `/usr/app/dlt`), bind-mounted read-only from the host project source | +| `DLT_ENTRYPOINT` | yes | Filename (relative to `DLT_PROJECT_DIR`) that `entrypoint.sh` runs with `python` (default `pipeline.py`) | +| `DLT_PIPELINES_DIR` | baked in | Where dlt persists pipeline working state (schema history, incremental load cursors) across runs (default `/usr/app/dlt_state`, a writable volume — the container filesystem is otherwise read-only) | +| `DLT_DATASET_NAME` / `DLT_PIPELINE_NAME` | no | Passed through for the pipeline script to read via `os.environ` (see `workdirs/dlt/pipeline.py`); dlt itself does not read these automatically | + +### Volumes + +| Path | Purpose | +| --- | --- | +| `$DLT_PROJECT_DIR` | Pipeline source (the script named by `DLT_ENTRYPOINT`) — mounted **read-only** | +| `$DLT_PIPELINES_DIR` | dlt's local pipeline working directory (schema, state, incremental load cursors) — must persist across runs for incremental sources to work | + +## Why no destination beyond Postgres yet? + +`images/dlt/requirements.txt` pins `dlt[postgres]`, matching the only +warehouse module CDS currently ships (`modules/warehouse/postgres`). Add the +relevant `dlt[]` extra and a new `DESTINATION____...` env +var if/when another warehouse module is added. + +## Source + +- Dockerfile and supporting files: [`images/dlt`](https://github.com/RonaldHensbergen/composable-data-stack/tree/main/images/dlt) +- Module definition: [`modules-experimental/ingestion/dlt/module.yaml`](https://github.com/RonaldHensbergen/composable-data-stack/tree/main/modules-experimental/ingestion/dlt) +- Sample pipeline: [`workdirs/dlt/pipeline.py`](https://github.com/RonaldHensbergen/composable-data-stack/tree/main/workdirs/dlt/pipeline.py) +- Issues and contributions: [RonaldHensbergen/composable-data-stack](https://github.com/RonaldHensbergen/composable-data-stack/issues) diff --git a/images/dlt/entrypoint.sh b/images/dlt/entrypoint.sh new file mode 100644 index 00000000..833044cc --- /dev/null +++ b/images/dlt/entrypoint.sh @@ -0,0 +1,13 @@ +#!/bin/sh +set -eu + +# Required by the module's compose environment (sourced from the consumed +# sql-database contract via ${bindings.*}). dlt's postgres destination reads +# this exact variable name itself; see +# https://dlthub.com/docs/dlt-ecosystem/destinations/postgres. +: "${DESTINATION__POSTGRES__CREDENTIALS:?DESTINATION__POSTGRES__CREDENTIALS is required}" +: "${DLT_PROJECT_DIR:?DLT_PROJECT_DIR is required}" +: "${DLT_ENTRYPOINT:?DLT_ENTRYPOINT is required}" + +cd "$DLT_PROJECT_DIR" +exec python "$DLT_ENTRYPOINT" diff --git a/images/dlt/requirements.txt b/images/dlt/requirements.txt new file mode 100644 index 00000000..53886920 --- /dev/null +++ b/images/dlt/requirements.txt @@ -0,0 +1,3 @@ +# dlt's postgres destination extra pulls in psycopg2-binary and the +# SQLAlchemy-based helpers dlt needs to load into Postgres. +dlt[postgres]==1.30.0 diff --git a/modules-experimental/ingestion/dlt/README.md b/modules-experimental/ingestion/dlt/README.md new file mode 100644 index 00000000..04bf7b27 --- /dev/null +++ b/modules-experimental/ingestion/dlt/README.md @@ -0,0 +1,53 @@ +# dlt (data load tool) + +Experimental ingestion module that runs a bind-mounted Python pipeline +script with [dlt](https://dlthub.com/) as a one-shot job, loading data into +a consumed `sql-database` contract (e.g. the `postgres` module). + +## Purpose + +Fills the ingestion gap in CDS's module set (warehouse, orchestration, bi, +secrets, cache): a provider-neutral way to run extract-load pipelines +against the warehouse without hardcoding which warehouse module is used. +The bundled `workdirs/dlt/pipeline.py` is a placeholder; replace it with a +real source (REST API, files, another database, etc.) when adopting this +module in a profile. + +## Known limitations + +- Experimental (`productionSuitable: false`): the module runs, but the + module/config shape may still change before it stabilizes. +- One-shot job only — dlt itself is a library, not a long-running server, + so there is no health check and `restart: "no"`. Scheduling repeated runs + (cron, an external trigger, or a future Dagster integration) is outside + this module's scope; see the non-goals in issue #589. +- Only the `postgres` destination is wired up (`dlt[postgres]` in + `images/dlt/requirements.txt`), matching the only warehouse module CDS + currently ships. +- No `provides` contract: since the job produces no long-running service, + there is nothing for another module to consume yet. A future + orchestration-triggered "ingestion" contract (so Dagster can consume and + trigger this pipeline) is tracked as follow-up work in issue #589, not + implemented here. + +## Upstream documentation + +- [dlt documentation](https://dlthub.com/docs/intro) +- [dlt postgres destination](https://dlthub.com/docs/dlt-ecosystem/destinations/postgres) + +## Configuration notes + +- `pipeline.hostPath`/`pipeline.containerPath` bind-mount the pipeline + project directory read-only, mirroring the `dbt` module's `project.*` + pattern. +- `entrypointScript` names the file (relative to `pipeline.containerPath`) + that `images/dlt/entrypoint.sh` runs with `python`. +- `destinationDatabase.contractRef` binds to a `sql-database` contract + (e.g. `postgres.sql-database`); its `connectionUri` is passed through as + `DESTINATION__POSTGRES__CREDENTIALS`, the exact environment variable name + dlt's postgres destination reads itself. +- `destinationDataset`/`pipelineName` are passed through as + `DLT_DATASET_NAME`/`DLT_PIPELINE_NAME` for the pipeline script to read; + dlt does not read these automatically, unlike the credentials variable. +- dlt's local pipeline state (schema history, incremental load cursors) + persists on the `dlt-state` volume across runs. diff --git a/modules-experimental/ingestion/dlt/module.yaml b/modules-experimental/ingestion/dlt/module.yaml new file mode 100644 index 00000000..3c2a44b6 --- /dev/null +++ b/modules-experimental/ingestion/dlt/module.yaml @@ -0,0 +1,132 @@ +# yaml-language-server: $schema=../../../cli/resources/module.schema.json +apiVersion: cds/v1alpha1 +kind: Module + +metadata: + name: dlt + category: ingestion + version: "0.1.0" + productionSuitable: false + displayName: dlt (experimental) + description: > + Experimental dlt (data load tool) ingestion module. Runs a bind-mounted + Python pipeline script as a one-shot job that extracts from a source and + loads into a consumed sql-database contract (e.g. the postgres module). + See images/dlt/README.md for configuration details. + +spec: + runtime: + type: container + service: + name: dlt + ports: [] + networks: + - default + + configSchema: + type: object + additionalProperties: false + required: + - destinationDatabase + properties: + pipeline: + type: object + additionalProperties: false + default: {} + properties: + hostPath: + type: string + minLength: 1 + default: ./workdirs/dlt + containerPath: + type: string + minLength: 1 + pattern: "^/" + default: /usr/app/dlt + + entrypointScript: + type: string + minLength: 1 + pattern: "^(?!\\.\\.?$)[A-Za-z0-9_.-]+$" + default: pipeline.py + description: > + Filename (relative to config.pipeline.containerPath) that + entrypoint.sh runs with `python`. Replace workdirs/dlt/pipeline.py + with a real extract-load pipeline; this only needs to be a valid + Python script that dlt can execute. + + destinationDatabase: + type: object + additionalProperties: false + required: + - contractRef + properties: + contractRef: + type: string + pattern: "^[a-z0-9-]+\\.[a-z0-9-]+$" + + destinationDataset: + type: string + minLength: 1 + default: raw + description: > + dlt "dataset_name" — the schema/dataset the pipeline loads into at + the consumed destination. Exposed to the pipeline script as the + DLT_DATASET_NAME environment variable. + + pipelineName: + type: string + minLength: 1 + default: cds_dlt_pipeline + description: > + dlt "pipeline_name" — identifies this pipeline's local state + (schema history, incremental load cursors) on the dlt-state + volume. Exposed to the pipeline script as the DLT_PIPELINE_NAME + environment variable. + + consumes: + - name: destination-database + contract: + kind: sql-database + required: true + mappedFrom: spec.config.destinationDatabase + + implementation: + kind: docker-compose + compose: + services: + dlt-run: + build: + context: ../../../ + dockerfile: images/dlt/Dockerfile + image: local/dlt:custom + restart: "no" + init: true + read_only: true + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + pids_limit: 256 + tmpfs: + - /tmp:rw,noexec,nosuid,nodev,uid=999,gid=999,mode=1777 + - /home/dlt/.dlt:rw,noexec,nosuid,nodev,uid=999,gid=999,mode=0700 + hostname: "${service.host}" + healthcheck: + disable: true + volumes: + - type: bind + source: "${config.pipeline.hostPath}" + target: "${config.pipeline.containerPath}" + read_only: true + - dlt-state:/usr/app/dlt_state + environment: + DESTINATION__POSTGRES__CREDENTIALS: "${bindings.destination-database.connectionUri}" + DLT_DATASET_NAME: "${config.destinationDataset}" + DLT_PIPELINE_NAME: "${config.pipelineName}" + DLT_PIPELINES_DIR: /usr/app/dlt_state + DLT_PROJECT_DIR: "${config.pipeline.containerPath}" + DLT_ENTRYPOINT: "${config.entrypointScript}" + + volumes: + dlt-state: {} diff --git a/tests/fixtures/signed-images.json b/tests/fixtures/signed-images.json index b30a53f2..4924e3ee 100644 --- a/tests/fixtures/signed-images.json +++ b/tests/fixtures/signed-images.json @@ -35,6 +35,13 @@ "signed": true, "provenanceAttested": true, "sbomAttested": true + }, + "cds-dlt": { + "repository": "ghcr.io/ronaldhensbergen/cds-dlt", + "digest": "sha256:ac4fd4d4a1cb9b826b6f59c901aa8938bf09fc41ae892c111eb577dc91908a4c", + "signed": true, + "provenanceAttested": true, + "sbomAttested": true } } } diff --git a/tests/test_dlt_postgres_wiring.py b/tests/test_dlt_postgres_wiring.py new file mode 100644 index 00000000..9aa7afc4 --- /dev/null +++ b/tests/test_dlt_postgres_wiring.py @@ -0,0 +1,145 @@ +""" +Committed dlt -> postgres binding test. + +Proves, via the real plan/render pipeline (not an uncommitted scratch +profile), that the experimental dlt module's `destinationDatabase.contractRef` +binding to a postgres module's `sql-database` contract resolves all the way +through to the rendered `dlt-run` service's `DESTINATION__POSTGRES__CREDENTIALS` +environment variable -- i.e. the consumed `connectionUri` is not left as an +unresolved `${bindings...}` expression, and the password stays an unresolved +`${CDS_*}` placeholder for Docker Compose to fill in at runtime (per repo +convention: secrets never resolve into plans or rendered Compose). +""" +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import yaml + +from cli.planner import build_plan +from cli.renderer import render_compose +from cli.validator import validate_profile + +_REPO_ROOT = Path(__file__).resolve().parent.parent + + +class DltPostgresWiringTest(unittest.TestCase): + """Wires the real dlt and postgres modules together and renders them.""" + + def _write_profile(self, profile_dir: Path) -> Path: + profile = { + "apiVersion": "cds/v1alpha1", + "kind": "Profile", + "metadata": {"name": "dlt-postgres-wiring", "environment": "local"}, + "spec": { + "runtime": {"type": "docker-compose"}, + "modules": [ + { + "id": "postgres", + "source": "modules/warehouse/postgres", + "version": "0.1.0", + "enabled": True, + "config": { + "database": "analytics", + "username": "analytics", + "passwordFrom": "secrets.analytics_db_password", + "superuserPasswordFrom": "secrets.postgres_superuser_password", + "port": 5432, + }, + }, + { + "id": "dlt", + "source": "modules-experimental/ingestion/dlt", + "version": "0.1.0", + "enabled": True, + "dependsOn": ["postgres"], + "config": { + "destinationDatabase": { + "contractRef": "postgres.sql-database", + }, + }, + }, + ], + "secrets": { + "provider": {"type": "env"}, + "values": { + "analytics_db_password": { + "env": "CDS_ANALYTICS_DB_PASSWORD", + "required": True, + }, + "postgres_superuser_password": { + "env": "CDS_POSTGRES_SUPERUSER_PASSWORD", + "required": True, + }, + }, + }, + }, + } + + profile_file = profile_dir / "profile.yaml" + profile_file.write_text(yaml.safe_dump(profile), encoding="utf-8") + return profile_file + + def test_destination_postgres_credentials_env_var_resolves(self): + with tempfile.TemporaryDirectory() as root: + profile_dir = Path(root) + profile_file = self._write_profile(profile_dir) + + env_file = profile_dir / ".env" + env_file.write_text( + "CDS_ANALYTICS_DB_PASSWORD=analytics_testpass\n" + "CDS_POSTGRES_SUPERUSER_PASSWORD=superuser_testpass\n", + encoding="utf-8", + ) + + with mock.patch.dict( + "os.environ", {"CDS_MODULE_PATH": str(_REPO_ROOT)}, clear=False + ): + diagnostics = validate_profile(str(profile_file)) + self.assertEqual( + [d for d in diagnostics if d.level == "error"], [], + msg=f"unexpected validation errors: {diagnostics}", + ) + + plan, plan_diags = build_plan(str(profile_file), env_file=str(env_file)) + self.assertIsNotNone(plan) + self.assertEqual( + [d for d in plan_diags if d.level == "error"], [], + msg=f"unexpected plan errors: {plan_diags}", + ) + + dlt_entry = next(m for m in plan["modules"] if m["id"] == "dlt") + bound_uri = dlt_entry["consumes"]["destination-database"]["contract"]["spec"][ + "connectionUri" + ] + # The binding must be fully resolved to the producer's + # rendered connection string (host/user/db from config, + # password still a runtime placeholder) -- not left as the + # raw "${bindings.destination-database.connectionUri}" + # expression. + self.assertNotIn("${bindings.", bound_uri) + self.assertEqual( + bound_uri, + "postgresql://analytics:${CDS_ANALYTICS_DB_PASSWORD}@postgres:5432/analytics", + ) + + output, render_diags = render_compose(plan, env_file=str(env_file)) + self.assertEqual( + [d for d in render_diags if d.level == "error"], [], + msg=f"unexpected render errors: {render_diags}", + ) + + compose = yaml.safe_load(output) + dlt_service = compose["services"]["dlt-run"] + credentials = dlt_service["environment"]["DESTINATION__POSTGRES__CREDENTIALS"] + + self.assertNotIn("${bindings.", credentials) + self.assertEqual( + credentials, + "postgresql://analytics:${CDS_ANALYTICS_DB_PASSWORD}@postgres:5432/analytics", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/workdirs/dlt/pipeline.py b/workdirs/dlt/pipeline.py new file mode 100644 index 00000000..f84d6490 --- /dev/null +++ b/workdirs/dlt/pipeline.py @@ -0,0 +1,36 @@ +"""Minimal example dlt pipeline for the CDS `dlt` ingestion module. + +Replace `sample_source` below with a real extractor (a REST API, files, a +source database, etc.) -- this only demonstrates that the pipeline can +connect to and load into the destination configured by the module's +consumed sql-database contract. + +DESTINATION__POSTGRES__CREDENTIALS, DLT_PIPELINE_NAME, and DLT_DATASET_NAME +are set by the dlt module's compose environment (see +modules-experimental/ingestion/dlt/module.yaml and images/dlt/entrypoint.sh); +dlt itself reads DESTINATION__POSTGRES__CREDENTIALS directly, this script +only needs to read the pipeline/dataset names. +""" +import os + +import dlt + + +@dlt.resource(name="ping") +def sample_source(): + """Trivial resource so the pipeline is runnable out of the box.""" + yield {"id": 1, "message": "hello from the cds dlt module"} + + +def main() -> None: + pipeline = dlt.pipeline( + pipeline_name=os.environ.get("DLT_PIPELINE_NAME", "cds_dlt_pipeline"), + destination="postgres", + dataset_name=os.environ.get("DLT_DATASET_NAME", "raw"), + ) + load_info = pipeline.run(sample_source()) + print(load_info) + + +if __name__ == "__main__": + main()