Metadata-Version: 2.5
Name: leeghwater
Version: 0.1.1
Summary: Your dlt pipeline, the same on your laptop and in a Databricks job
Project-URL: Homepage, https://github.com/kostavo-oss/tools
Project-URL: Documentation, https://kostavo-oss.github.io/tools/leeghwater/
Project-URL: Repository, https://github.com/kostavo-oss/tools
Project-URL: Issues, https://github.com/kostavo-oss/tools/issues
Project-URL: Changelog, https://github.com/kostavo-oss/tools/blob/main/packages/leeghwater/CHANGELOG.md
Author: Misja Pronk
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: cli,databricks,dlt,dlthub,ingestion,secrets
Classifier: Development Status :: 3 - Alpha
Classifier: Environment :: Console
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.11
Requires-Dist: databricks-sdk>=0.50
Requires-Dist: dlt<2,>=1.30
Requires-Dist: typer>=0.17.5
Provides-Extra: keyvault
Requires-Dist: azure-identity>=1.15; extra == 'keyvault'
Requires-Dist: azure-keyvault-secrets>=4.8; extra == 'keyvault'
Description-Content-Type: text/markdown

# leeghwater

**Your dlt pipeline, the same on your laptop and in a Databricks job.**
One command to run it, its secrets from a secret scope, and the things Databricks asks for
kept out of your code.

[![ci](https://github.com/kostavo-oss/tools/actions/workflows/ci.yml/badge.svg)](https://github.com/kostavo-oss/tools/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/leeghwater.svg)](https://pypi.org/project/leeghwater/)
[![Python](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://www.python.org/downloads/)
[![License: Apache-2.0](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](https://github.com/kostavo-oss/tools/blob/main/packages/leeghwater/LICENSE)

> **Early.** It has run as a serverless job and from a laptop against a real workspace;
> what was tried and what was not is under [Tried, and not yet](#tried-and-not-yet).

leeghwater is a package your [dlt](https://dlthub.com) project imports. Your sources stay
ordinary dlt code. leeghwater gives the project a command line, and for each run it makes
the dlt pipeline from what the run names: a local file on a laptop, the schema your bundle
deployed in a job.

## Why

A dlt pipeline on Databricks tends to run only on a cluster. Every change is a deploy and
a wait, and the secrets are in one place on a laptop and another in a job. So nobody runs
the pipeline locally, and finding a typo takes ten minutes. With leeghwater the same
command runs the pipeline on a laptop into a local file and unchanged in a job, with its
secrets from one place.

## A project is two files

The app, which is the project's console script:

```python
# src/my_ingest/cli.py
from leeghwater import App

app = App(pipelines="my_ingest.pipelines", secret_scopes=["ingest"])
```

```toml
# pyproject.toml
[project.scripts]
ingest = "my_ingest.cli:app"
```

And a pipeline: a function marked `@pipeline`, anywhere in the package the app names.

```python
# src/my_ingest/pipelines/github.py
from datetime import date

import dlt
import leeghwater
from leeghwater import pipeline


@dlt.source
def github_source(since: date, access_token: str = dlt.secrets.value): ...


@pipeline
def github(since: date = date(2024, 1, 1)):
    """Load GitHub issues."""
    p = leeghwater.create_pipeline("github")
    return leeghwater.run(p, github_source(since=since))
```

`create_pipeline` and `run` stand in for `dlt.pipeline()` and `pipeline.run()`, and a
pipeline never calls those two itself. Where a run loads, into which schema and through
which staging schema is the run's to say, not the code's: `create_pipeline` sets them on
the dlt destination it makes, and a plain `dlt.pipeline()` gets none of it. What you get
back is dlt's own pipeline and dlt's own load info, and anything else the two dlt calls
take is passed on.

```sh
ingest list                                  # the pipelines and their parameters
ingest run github --since 2024-06-01         # run one
ingest doctor                                # what would be set up, checked, nothing run
```

## Start a project

[data-product-template](https://github.com/kostavo-oss/data-product-template) is the way to start: a copier
template for a data product — dlt to land the data, run by leeghwater; dbt to shape it; one
job; the schemas as bundle resources; every task behind `mise run`.

```sh
uvx copier copy gh:kostavo-oss/data-product-template my-product
```

To add leeghwater to a project you have: `uv add leeghwater`, write the two files above, and
pass the schemas' deployed names from the bundle as shown under [In a job](#in-a-job).

## Where a run loads

| Where you run it | It loads into | Secrets come from |
|---|---|---|
| A laptop | a local DuckDB file, `<pipeline>.duckdb` | the environment, `.dlt/secrets.toml` |
| A laptop, with `--profile <name>` | that workspace, as you | the same, then the secret scopes |
| Databricks | this workspace, as the run's identity | the environment, then the secret scopes |

The first row needs nothing but Python and `dlt[duckdb]`. It tries your source, not the
destination: DuckDB ignores what is Databricks' own, such as table properties.

A project that loads somewhere else says so once, in the app:
`App(..., destination=..., local_destination=...)`.

## The options of every run

A pipeline's own parameters are options of its command: `since` above is `--since`, with
the type the function gives it. Text, whole and decimal numbers, dates and times in ISO
format, and yes or no as a word (`--full true`), so a bundle variable can set it.

Every run also takes these:

| Option | What it sets |
|---|---|
| `--profile <name>` | Load into a workspace from a laptop, with a profile from `~/.databrickscfg`. Refused on Databricks. |
| `--catalog <name>` | The catalog to load into. |
| `--schema <name>` | The schema to load into, used exactly as given. |
| `--staging-schema <name>` | The schema dlt stages through, by its whole name. |
| `--warehouse <id>` | The SQL warehouse to load through. |
| `--secret-scope <name>` | A secret scope to read, in place of the app's. May be repeated. |
| `--secret <name>=<scope>/<key>` | Where one secret is, when it isn't under dlt's name. May be repeated. |
| `--key-vault <url>` | An Azure Key Vault to read, after the scopes. May be repeated. |
| `--limit <n>` | At most `n` pages from each resource; `0`, the default, for all. |
| `--log-level <level>` | `DEBUG`, `INFO`, `WARNING`, `ERROR` or `CRITICAL`, for dlt. |
| `--set <key>=<value>` | Any dlt config value, by dlt's name: `--set sources.github.page_size=50`. |

The schema, staging schema, catalog and warehouse are set on the destination that
`create_pipeline` makes, in the call, which is what dlt takes first. `--set` and
`--log-level` go to dlt as config, the way an environment variable does. Nothing in dlt is
patched.

A pull-request environment runs with `--limit 1`: enough for dlt to make every table with
its real columns. Not with nothing: dlt takes a table's columns from the rows it sees, so a
source that gives no rows makes no table.

## In a job

The same command is the job's entry point. Nothing in the task names leeghwater.

```yaml
python_wheel_task:
  package_name: my-ingest
  entry_point: ingest
  parameters:
    - run
    - github
    - --catalog
    - ${var.catalog}
    - --schema
    - ${resources.schemas.github.name}
    - --staging-schema
    - ${resources.schemas.github_staging.name}
    - --warehouse
    - ${var.warehouse_id}
```

### Why the schemas are passed by reference

A bundle names your schemas, and a target may change the name it deploys: a prefix, or a
suffix per developer, such as `github__anna`. Left alone, dlt gets this wrong twice:

- it normalizes a dataset name, and `github__anna` becomes `github_anna`: a schema nobody
  deployed;
- it derives its staging dataset as `<name>_staging`, so `github__anna_staging`, while the
  bundle deployed `github_staging__anna`. dlt then makes a schema of its own, outside the
  bundle's grants and its cleanup.

`--schema` turns the normalizing off for the run, and `--staging-schema` gives dlt the
staging schema's whole name. On Databricks a run that names a schema and no staging schema
stops, and says which line of the job is missing. That holds for a pipeline that never
merges too: a staging schema that is there and unused costs nothing, one that is missing
costs a schema nobody governs.

### A run ends the way it went

A run prints what was set up, then a report of the load. It ends with an error when:

- the pipeline raises;
- a load job failed. dlt raises on that by itself; a project that turned that off
  (`load.raise_on_failed_jobs`) still gets a failed run;
- a run into Databricks names no warehouse. dlt would otherwise load through the first
  warehouse on the workspace's list;
- the pipeline made no load with `leeghwater.run()`: a plain `dlt.pipeline()` loads
  wherever dlt's own config says, with no schema, catalog or limit from the run.

A serverless job retries a failed task once by itself unless the task sets
`disable_auto_optimization: true`: set it, so a load is retried when you say so, with
`max_retries`.

## Secrets

A source asks dlt for a secret: `access_token: str = dlt.secrets.value`. dlt looks in the
environment, then in `.dlt/secrets.toml`, then in the secret scopes the app or the run
names. Three ways to give it one, and when to use each:

1. **By dlt's name, in a scope.** The path dlt asks for, joined with `-`:

   ```sh
   databricks secrets put-secret ingest sources-github-access_token
   ```

   The default. One secret per value, and no code.

2. **A whole section in one secret.** A secret named `sources-github` (or `sources`,
   `destination-<name>`, `destination`, `dlt_secrets_toml`) is read as a piece of
   `secrets.toml`. It must spell out its own heading:

   ```toml
   [sources.github]
   access_token = "..."
   ```

   Without the heading its values are found as every source's. Use it when a source has
   several secrets that change together.

3. **A secret that exists under another name.** Point at it, as `<scope>/<key>`:

   ```python
   app = App(..., secrets={"sources.github.access_token": "platform-shared/github-token"})
   ```

   or per run, so dev and prod can differ in the bundle:
   `--secret sources.github.access_token=platform-shared/github-token`.
   Use it when somebody else fills the scope.

4. **An Azure Key Vault**, for a team whose secrets live there and not in a scope. Needs
   the extra, `uv add "leeghwater[keyvault]"`, and a vault named in the app or per run:

   ```python
   app = App(..., key_vaults=["https://kv-ingest.vault.azure.net"])
   ```

   A vault is asked after the scopes. Its secrets are named like a scope's, with dashes
   where a Key Vault allows no underscore: `sources-github-access-token`. Whoever
   `DefaultAzureCredential` finds reads it: `az login` on a laptop; a service principal's
   variables or a managed identity elsewhere. A Databricks job has no identity in Azure by
   itself, so there a vault is reachable only through a service principal whose secret
   comes from somewhere else — in practice a Databricks scope — and only when the job's
   network reaches `vault.azure.net`. A Key Vault-backed Databricks scope needs none of
   this: it is a scope, read as in 1.

A scope or a vault is listed once in a run and only the secrets that are there are read.
Only values a source marks as secret are looked up. No value is printed, logged, written to
a file or put in the environment. To make, change or rotate a secret, use the Databricks
CLI or the Azure portal.

In a job a secret is read through `dbutils`, so Databricks redacts it: if your own code
prints one, the job's output shows `[REDACTED]`.

## In a notebook, or without an app

```python
import leeghwater

setup = leeghwater.prepare(secret_scopes=["ingest"], catalog="main", warehouse="<id>")
print("\n".join(setup.lines()))

p = leeghwater.create_pipeline("github", schema="github")
leeghwater.run(p, github_source())
```

`prepare()` is everything the app does before a run, and returns what it decided;
`create_pipeline` and `run` go by the last `prepare()`, or by a `setup=` you pass them.
Importing `leeghwater` changes nothing and does not import dlt.

## Commands of your own

```python
@app.command()
def backfill(day: str):
    """Load one day again."""
    ...
```

`ingest backfill 2024-06-01` is prepared as a pipeline is, with the app's own settings, and
finds the same secrets.

## Troubleshooting

- **Deleting the DuckDB file does not start a pipeline over.** dlt keeps a pipeline's state
  outside the database, under `~/.dlt/pipelines/<name>/`. Delete that folder too.
- **`import dlt` gives Delta Live Tables.** Databricks has a module of its own named `dlt`.
  In a serverless job a plain `import dlt` gets the right one. In a notebook or on a
  cluster, call `leeghwater.prepare()` before `import dlt`. The first lines of a run say
  what was done about the name.
- **The pipeline and its schema have the same name, and DuckDB refuses.** DuckDB names its
  database after the file, `<pipeline>.duckdb`, and can't tell a schema of that name from
  it. Locally, give the schema another name or leave it out.
- **A load from a job is refused a connection to `…storage.cloud.databricks.com`.** dlt
  uploads its files to a volume before it copies them into a table. A workspace whose
  serverless compute has a limited outbound network refuses that upload.
- **`ingest doctor`** says where a run would load and checks the sign-in, each secret scope
  and the warehouse, without running anything. `--json` for a machine.

## Tried, and not yet

On 2026-10-07 a project made by `leeghwater init` was deployed to a workspace as a bundle
and run, and the same project was run from a laptop with a profile.

Held, for real:

- a bundle made from the former `leeghwater init` validates, deploys and destroys, and the
  job's entry point is the project's own command;
- the job is told the deployed names of the bundle's schemas, and finds the `config.toml`
  its wheel carries;
- a pipeline that raises fails its task;
- a secret scope is read as the job's identity and from a laptop, and a secret is redacted
  in the job's output;
- from a laptop, a load into Unity Catalog; and with schemas named with two underscores, a
  load into exactly those, with no third schema made.

Not yet:

- **a full load from a serverless job.** The workspace it was tried on limits the outbound
  network of serverless compute, and dlt's upload to storage was refused. Everything before
  the upload ran;
- a notebook and a classic cluster, which is where the name `dlt` is said to bite;
- the job's `ingest` task with leeghwater installed from PyPI, which 0.1.0 makes possible.

## When not to use it

- **Lakeflow Connect has a connector for your source** and it does what you need.
- **dlt runs somewhere else** — Airflow, GitHub Actions — and only loads into Databricks.
  dlt does that alone.
- **The data needs Spark to move it.** dlt is Python on one machine, and leeghwater
  doesn't change that.

leeghwater is a library for one kind of code, dlt ingestion. It will not:

- deploy, schedule, or order one pipeline after another. That is a job's, and a bundle's;
- add a file format for pipelines, or patch anything in dlt;
- make, change or rotate secrets. That is the Databricks CLI's, and caland's;
- change tables, or set table properties and clustering. dlt's own `databricks_adapter`
  does the second, in your source;
- change anything on a cluster: no init script, no file outside the run's own
  directories.

## When you no longer need it

When dlt reads Databricks secret scopes by itself and runs in a job with nothing around
it — the right `dlt` imported, the bundle's schema names used as they are — use dlt
alone. Today dlt has no provider for secret scopes, and its Databricks
page gives workarounds for the import.

## Where it fits

leeghwater is one of the [Kostavo tools](https://github.com/kostavo-oss/tools): small tools for the ugly gaps on
Databricks, one gap each. Its gap is the pipeline that only runs on a cluster: every
change a deploy and a wait, and the secrets in one place on a laptop and another in a job.
Each works alone. Kostavo is the company behind them: it builds
[a governance platform for Databricks workspaces](https://kostavo.com), and the tools are
complete without it.

Community project, not affiliated with or endorsed by Databricks or dltHub.

## Named after

Jan Adriaanszoon Leeghwater — the millwright who drained the Beemster with windmills, dry in 1612. He moved water from where it lay to where it was wanted.

## License

[Apache-2.0](https://github.com/kostavo-oss/tools/blob/main/packages/leeghwater/LICENSE).
