Metadata-Version: 2.4
Name: ikietl
Version: 1.1.1
Summary: Table-driven ETL Swiss-knife with CLI and Python API
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: adlfs>=2024.6
Requires-Dist: fsspec>=2024.6
Requires-Dist: gcsfs>=2024.6
Requires-Dist: openpyxl>=3.1
Requires-Dist: pandas>=2.2
Requires-Dist: psycopg2-binary>=2.9
Requires-Dist: pyarrow>=16.0
Requires-Dist: pydantic>=2.6
Requires-Dist: pymysql>=1.1
Requires-Dist: pyyaml>=6.0
Requires-Dist: rich>=13.7
Requires-Dist: s3fs>=2024.6
Requires-Dist: simpleeval>=1.0
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: tomli>=2.0; python_version < '3.11'
Requires-Dist: typer>=0.12
Provides-Extra: polars
Requires-Dist: polars>=1.0; extra == 'polars'
Description-Content-Type: text/markdown

# Iki-ETL-Table-Driven (`ikietl`)

A single, table-driven ETL Swiss-army-knife. Every stage — **extract →
transform → validate → load** — is described as data (arrays of stage-specs)
against any file location or any database, and a small facade hides the
pipeline DAG underneath.

![Image_Cover](assets/image.png)

**This is not just a config-file tool.** The same pipeline definition can
come from a YAML/TOML/JSON file **or** be built entirely in Python as a plain
`dict` and handed to `build_pipeline_config(...)` — no file on disk required.
The CLI (`etl`) is a thin wrapper over a Python package (`ikietl`) that you
can import and drive directly: construct configs programmatically, inspect
`PipelineConfig` objects, call `run_pipeline()`/`validate()`/`status()` from
your own scripts, orchestrator, or test suite. See
[Python API](#python-api) below.

```
etl run | dry-run | validate | status | audit | clean | init
```

- **Table-driven** — the pipeline is arrays of declarative stage-specs, not code
- **Config file _or_ Python dict** — write `ikietl.yaml` and use the CLI, or call `build_pipeline_config({...})` from Python and skip files entirely
- **Universal source** — any DB (`SQLAlchemy`) + any storage backend (`fsspec`) + any file format (registry), read/written with either **pandas** or **polars**
- **Declarative transforms** — ops for the common cases (rename, filter, join, dedup, …), plus a restricted expression evaluator for row-level logic
- **Schema-validated config** — `Pydantic` catches structural mistakes before a byte is read, whether the config came from a file or a dict
- **Examples-first** — a gallery of runnable ETL scenarios, each with its own workspace under `examples/`

> **Before you build on this:** an independent audit of this codebase found two
> advertised features that do **not** currently work (`op: custom` transforms,
> and SQL `mode: upsert`), plus a couple of path-resolution edge cases. Everything
> else described below has been verified by running the actual code.

---

## Install

```bash
pip install ikietl
```

One package, no extras to remember for the default setup — it ships with
everything the tool supports out of the box: cloud filesystems (`s3fs`,
`gcsfs`, `adlfs`), Postgres/MySQL drivers (`psycopg2-binary`, `pymysql`), and
Excel (`openpyxl`). All file I/O runs on **pandas** by default.
(There is currently no bundled SQL Server driver — add `pyodbc` or `pymssql`
yourself and pass the matching SQLAlchemy connection string if you need it.)

If you also want the **polars** I/O engine (see
[I/O engines: pandas & polars](#io-engines-pandas--polars) below), install
the optional extra:

```bash
pip install "ikietl[polars]"
```

`ikietl` works exactly the same without it — polars is opt-in per
extract/load step, and you only need it installed if a step actually
requests `engine: polars`.

Editable/dev install:

```bash
pip install -e .
pip install pytest   # only needed to run the test suite
```

---

## 60-second quickstart

```bash
etl init                            # scaffold ikietl.yaml + work/logs/state/audit/tmp/transforms
etl validate                        # config schema + data-contract checks (extract -> transform -> validate, no load)
etl dry-run                         # same stages as validate
etl run                             # execute the full pipeline, including load
etl status                          # show last run's state (timestamp, dataset row counts)
etl audit                           # show the lineage log (one line per stage per dataset)
etl clean                           # remove work/ and tmp/ (state/ is kept)
```

`etl` looks for `ikietl.yaml` / `ikietl.yml` / `ikietl.toml` / `ikietl.json` in
the current directory by default; pass `--config`/`-c` to point at a specific
file. All three formats parse into the same validated Pydantic model, so
every command below works identically regardless of which one you use.

### Try the example gallery

The repository ships with several runnable example scenarios under
`examples/`, each backed by a small Python script that seeds sample data,
writes a matching `ikietl.yaml`, and runs the pipeline through the real
engine (no network or database server required — SQL examples use an
in-memory/temp SQLite database):

```bash
python -c "from examples.basic_examples import run_database_etl_example; print(run_database_etl_example())"
python -c "from examples.basic_examples import run_cloud_storage_etl_example; print(run_cloud_storage_etl_example())"
```

Every function in `examples/basic_examples.py` (`run_database_etl_example`,
`run_incremental_etl_example`, `run_mixed_source_etl_example`,
`run_quality_monitoring_etl_example`, `run_event_stream_etl_example`,
`run_unstructured_data_etl_example`, `run_cloud_storage_etl_example`, and
several smaller `run_*_example` helpers) is self-contained and safe to call
directly — they're also exercised by `tests/test_examples.py`.

Each scenario also has a matching folder (e.g. `examples/database-etl/`)
containing the `ikietl.yaml` it produces plus `input/`, `output/`, `work/`,
`logs/`, `state/`, and `tmp/` sub-folders — inspect those after running the
example above to see exactly what the pipeline read and wrote.

> Note: there are currently no Docker Compose files in this repository for
> Postgres/MySQL/MinIO backends — the "database" and "cloud-storage" examples
> run entirely against local SQLite/filesystem stand-ins so they work with no
> setup. If you want to point them at a real Postgres/MySQL/S3 endpoint,
> supply your own connection string / bucket via the `conn` / `path` fields
> and (for cloud storage) the relevant `storage_options`.

---

## How a config file is structured

Every `ikietl.yaml` (or `.toml`/`.json`) has up to five top-level sections:

```yaml
meta: { ... } # required — name/version/description
runtime: { ... } # optional — work/log/state/tmp dirs, strict mode, parallelism
extract: [...] # required — where data comes from
transform: [...] # optional — how it's reshaped
validate: [...] # optional — data-contract checks
load: [...] # required — where results go
```

Datasets flow through the pipeline by **id**: each `extract` entry produces a
named dataset; each `transform` entry reads one named dataset (`input:`) and
produces a new one (its own `id:`); `validate` and `load` entries reference
any dataset by id. IDs must be unique across extract + transform outputs.

### `meta` — required

```yaml
meta:
  name: my-pipeline
  version: "1.0" # optional, defaults to "1.0"
  description: "..." # optional, defaults to ""
```

### `runtime` — optional, but read the caveat below

```yaml
runtime:
  work_dir: ./work # default: ./work
  log_dir: ./logs # default: ./logs
  state_dir: ./state # default: ./state
  tmp_dir: ./tmp # default: ./tmp
  strict: true # default: true — any validation failure aborts before load
  parallelism: 1 # default: 1 — >1 extracts run concurrently in a thread pool
```

⚠️ **Caveat:** when `runtime:` is present (even partially),
every path in it is resolved **relative to the config file's own directory**,
not your current working directory — this is what lets you run `etl` from
anywhere and still get consistent output locations. If you omit `runtime:`
**entirely**, the defaults above are used as literal, un-resolved relative
paths instead, which will resolve against whatever directory the process
happens to be launched from. **Recommendation: always include a `runtime:`
block, even if you're only overriding one field**, to get predictable,
config-relative paths.

`parallelism > 1` only affects the **extract** stage, and only when there's
more than one extract entry — each extractor runs in its own thread. Extract
results are still merged deterministically by dataset id afterwards, so
downstream transform/validate/load behavior is unaffected by ordering.

### `extract` — required, one or more sources

```yaml
extract:
  - id: sales_csv
    type: file
    path: "s3://my-bucket/incoming/sales_*.csv" # any fsspec URL, or a local path
    format: csv # optional — inferred from the extension if omitted
    options:
      delimiter: ","
      encoding: "utf-8"
      engine: pandas # optional — "pandas" (default) or "polars", see below
    glob: true # true = read + concat every file matching the path

  - id: customers_db
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/db" # any SQLAlchemy URL
    query: "SELECT * FROM customers WHERE updated_at > :last_run"
    params: { last_run: "${state.last_run}" } # bound as real SQL params — safe from injection
```

- `${env.VAR}` interpolates an environment variable. Missing variables raise
  a clear `ConfigError` at load time rather than failing deep inside a run.
- `${state.key}` interpolates a value from the **previous run's** state file
  (`state/last_run.json`) — the standard way to do incremental/watermark
  extraction (e.g. "only rows newer than last time I ran").
- `{date}` anywhere in a string is replaced with today's date (`YYYY-MM-DD`)
  before anything else is resolved — handy for date-partitioned paths.
- `glob: true` requires the path to actually match at least one file, or
  extraction fails loudly rather than silently returning nothing.

Local file paths (for both `extract` and `load`) are resolved relative to the
config file's directory automatically — a script running `etl` from any
other working directory still reads/writes in the same place relative to
`ikietl.yaml`.

### `transform` — optional, applied in step order

```yaml
transform:
  - id: clean_sales
    input: sales_csv
    steps:
      - op: rename
        map: { "Order ID": order_id, "Amount": amount }
      - op: cast
        column: amount
        type: decimal # int | float | string | datetime | decimal
      - op: filter
        expr: "amount > 0"
      - op: derive
        column: year
        expr: "year(order_date)"
      - op: select
        columns: [order_id, amount, year, customer_id]
      - op: join
        with: customers_db # another dataset id — must already exist
        on_column:
          customer_id # NOT `on:` — bare `on:` parses as the YAML 1.1
          # boolean key `True` in some YAML loaders, and
          # is only accepted here as a fallback alias
        type: left # left | right | inner | outer (passed to pandas.merge)
      - op: sort
        by: [order_date]
        ascending: true
      - op: dedup
        columns: [order_id]
```

`filter` and `derive` expressions run through [`simpleeval`](https://pypi.org/project/simpleeval/)
against the row's columns (plus the previous run's `state`, if any) — never
raw `eval()` — so a config file cannot execute arbitrary code through these
two ops. Whitelisted helper functions: `year()`, `month()`, `day()`,
`upper()`, `lower()`, `abs()`, `round()`.

Every transform op (`rename`, `select`, `drop`, `cast`, `filter`, `derive`,
`join`, `sort`, `dedup`) runs on a `pandas.DataFrame`, regardless of which
I/O `engine:` produced the dataset — see the next section for why.

**`op: custom` — supported.** The codebase supports declaring a custom
Python transform (`op: custom`) that points at a local `module.py` and a
`function` name. `module`/`function` may be supplied at the step top-level
or inside `args:`. Relative `module` paths are resolved relative to the
config file's directory. See `src/ikietl/transform/main.py` for the
implementation details.

### `validate` — optional, data-contract checks

```yaml
validate:
  - id: sales_quality
    input: clean_sales
    rules:
      - { type: not_null, columns: [order_id, amount] }
      - { type: unique, columns: [order_id] }
      - { type: range, column: amount, min: 0.01, max: 1000000 }
      - { type: row_count, min: 1 }
```

In `strict: true` mode (the default), any rule failure raises immediately and
aborts the run before `load` executes. In `strict: false` mode, a failure is
recorded to the audit/lineage log (`{"failed": true, "error": "..."}`) and
the pipeline continues to the next validation/load step instead of stopping.

### `load` — required, one or more destinations

```yaml
load:
  - id: archive_csv
    input: clean_sales
    type: file
    path: "./work/clean_sales.csv"
    format: csv
    options:
      delimiter: ";"
      engine: polars # optional — write with polars instead of pandas

  - id: warehouse
    input: clean_sales
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/warehouse"
    table: fact_sales
    mode: append # append | replace  — see caveat below
    transaction: true # wrap the write in a DB transaction (default: true)
```

- **File loads are atomic**: data is written to `path` + `.tmp` and then
  renamed into place, so a crash mid-write never leaves a half-written file
  at the real path.
- **SQL `mode`** supports `append`, `replace`, and `upsert` (key-based
  merge) — when `mode: upsert` is used you must set `key: [col1, col2]` on
  the load spec. `upsert` is implemented for Postgres and MySQL/MariaDB
  via a staging table + `INSERT ... ON CONFLICT` / `ON DUPLICATE KEY UPDATE`.
  `insert` is accepted as an alias for `append`.

---

## I/O engines: pandas & polars

File extract/load is a **table-driven registry**: one row per
`(format, engine)` pair, shared by every extractor and loader — adding a
format means adding a row, not a new code path. Two engines are supported,
selected per extract/load step with `options.engine`:

```yaml
options:
  engine: pandas # default — used automatically if you omit `engine:`
  # engine: polars   # opt-in — requires `pip install "ikietl[polars]"`
```

**Why both, and how they interoperate:** transform/validate always operate
on a `pandas.DataFrame` (that's the pipeline's internal contract), so an
`extract` step with `engine: polars` reads with polars — for its speed and
broader native-format support — and the result is converted to pandas
immediately afterwards via Arrow before it reaches your `transform:` steps.
A `load` step with `engine: polars` does the reverse: your pandas dataset is
converted right before writing, so the fast columnar polars writer is used.
Mixing engines across steps in the same pipeline is fine — a file written by
one engine can always be read back by the other.

### Format matrix

**pandas** (`engine: pandas`, the default) — mirrors the
[pandas I/O guide](https://pandas.pydata.org/docs/user_guide/io.html):

| Format           | Extension(s) inferred        | Read  |    Write     |
| ---------------- | ---------------------------- | :---: | :----------: |
| CSV              | `.csv`                       |  ✅   |      ✅      |
| TSV              | `.tsv`                       |  ✅   |      ✅      |
| Fixed-width text | `.txt`                       |  ✅   |      —       |
| JSON             | `.json`                      |  ✅   |      ✅      |
| JSON Lines       | `.jsonl`, `.ndjson`          |  ✅   |      ✅      |
| HTML             | `.html`, `.htm`              |  ✅   |      ✅      |
| XML              | `.xml`                       |  ✅   |      ✅      |
| Excel            | `.xlsx`, `.xls`              |  ✅   | ✅ (`.xlsx`) |
| OpenDocument     | `.ods`                       |  ✅   |      —       |
| Parquet          | `.parquet`, `.pq`            |  ✅   |      ✅      |
| Feather          | `.feather`, `.arrow`, `.ipc` |  ✅   |      ✅      |
| ORC              | `.orc`                       |  ✅   |      ✅      |
| HDF5             | `.h5`, `.hdf5`               | ✅ \* |    ✅ \*     |
| Stata            | `.dta`                       |  ✅   |      ✅      |
| SAS              | `.sas7bdat`                  |  ✅   |      —       |
| SPSS             | `.sav`                       | ✅ \* |      —       |
| Pickle           | `.pkl`, `.pickle`            |  ✅   |      ✅      |

Blank cells (`—`) match pandas' own IO table — those formats genuinely have
no writer upstream, not a gap in this registry.

**polars** (`engine: polars`, opt-in) — mirrors the
[polars I/O reference](https://docs.pola.rs/api/python/stable/reference/io.html):

| Format                | Read | Write |
| --------------------- | :--: | :---: |
| CSV                   |  ✅  |  ✅   |
| TSV                   |  ✅  |  ✅   |
| JSON                  |  ✅  |  ✅   |
| JSON Lines (ndjson)   |  ✅  |  ✅   |
| Parquet               |  ✅  |  ✅   |
| Feather / IPC / Arrow |  ✅  |  ✅   |
| Excel                 |  ✅  |  ✅   |
| Avro                  |  ✅  |  ✅   |

If a step requests a format the chosen engine doesn't support, extraction/load
fails fast with a clear `ValueError` naming the format, the engine, and every
format that engine _does_ support — not a confusing library-internal error.

\* **HDF5 and SPSS require a real filesystem path**, not an in-memory
stream — that's a limitation of the underlying `pytables`/`pyreadstat`
libraries, not this tool. Reading either format from a local `path:` works
exactly like any other format. Reading from a remote `path:` (`s3://`,
`gs://`, …) is still supported: the file is transparently staged to a local
temp file first, then read, then the temp file is removed. Writing HDF5 to a
remote `path:` is **not** supported — write it locally and upload separately.

Any reader/writer keyword arguments not covered by the first-class `options`
fields above (`delimiter`, `encoding`, `sheet`, `key`, …) can be passed
straight through via `options.read_kwargs` / `options.write_kwargs`, e.g.:

```yaml
options:
  engine: pandas
  read_kwargs: { dtype: { order_id: str } }
```

---

## Python API

Everything the CLI does is a thin wrapper around `ikietl`'s public API — you
can call the same functions directly from Python:

```python
from ikietl import (
    audit, build_pipeline_config, clean, init,
    load_config, run_pipeline, status, validate,
)
```

| Function                                                  | Signature                         | What it does                                                                                                                                                                                   |
| --------------------------------------------------------- | --------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `load_config(path)`                                       | `(str \| Path) -> PipelineConfig` | Parse + validate a YAML/TOML/JSON file into a `PipelineConfig`.                                                                                                                                |
| `build_pipeline_config(config_data, state_dir="./state")` | `(dict) -> PipelineConfig`        | Validate a plain Python `dict` into a `PipelineConfig` — **no file needed at all.**                                                                                                            |
| `run_pipeline(config, dry_run=False)`                     | `(PipelineConfig \| str) -> dict` | Run extract → transform → validate → (load unless `dry_run`). Accepts a `PipelineConfig` object or a path string. Returns `{"datasets": {id: row_count}, "state_dir": ..., "audit_dir": ...}`. |
| `validate(config)`                                        | `(PipelineConfig \| str) -> dict` | Same as `run_pipeline(config, dry_run=True)`.                                                                                                                                                  |
| `status(config)`                                          | `(PipelineConfig \| str) -> dict` | Read `state/last_run.json` for this config.                                                                                                                                                    |
| `audit(config)`                                           | `(PipelineConfig \| str) -> str`  | Read the raw `logs/audit/lineage.jsonl` contents.                                                                                                                                              |
| `clean(config)`                                           | `(PipelineConfig \| str) -> None` | Delete `work/`/`tmp/` contents (keeps `state/`).                                                                                                                                               |
| `init(name="my-pipeline")`                                | `(str) -> None`                   | Scaffold a starter `ikietl.yaml` + directories in the current directory.                                                                                                                       |

### Fully in-memory — no config file on disk

Because `build_pipeline_config` takes a plain `dict`, you can define, run, and
inspect a whole pipeline from a script or notebook without ever writing an
`ikietl.yaml`:

```python
from ikietl import build_pipeline_config, run_pipeline

cfg = build_pipeline_config({
    "meta": {"name": "in-memory-demo", "version": "1.0"},
    "runtime": {
        "work_dir": "./work", "log_dir": "./logs",
        "state_dir": "./state", "tmp_dir": "./tmp",
        "strict": True,
    },
    "extract": [{
        "id": "sales_csv", "type": "file",
        "path": "./input/sample_sales.csv", "format": "csv",
    }],
    "transform": [{
        "id": "clean_sales", "input": "sales_csv",
        "steps": [
            {"op": "cast", "column": "amount", "type": "decimal"},
            {"op": "filter", "expr": "amount > 0"},
        ],
    }],
    "validate": [{
        "id": "sales_quality", "input": "clean_sales",
        "rules": [{"type": "row_count", "min": 1}],
    }],
    "load": [{
        "id": "archive", "input": "clean_sales",
        "type": "file", "path": "./output/output.csv", "format": "csv",
    }],
})

result = run_pipeline(cfg)         # dry_run=True to skip the load stage
print(result["datasets"])          # {"clean_sales": 3}
```

This is exactly the pattern used throughout `examples/basic_examples.py`
(`run_extractors_example`, `run_pipeline_config_example`, etc.) — it's a
first-class, tested way to use the package, not a fallback.

### Calling the format registry directly

Outside of a full pipeline, the same table-driven registry that backs
`extract`/`load` is importable and usable on its own — handy in a notebook
or a one-off script:

```python
from ikietl.formats.registry import read, write, infer_format, available_formats

available_formats("polars")   # {"csv": True, "json": True, "parquet": True, ...}
infer_format("s3://bucket/events.jsonl")   # "jsonl"

with open("sales.parquet", "rb") as f:
    df = read("parquet", f, options={}, engine="pandas")

write(df, "csv", "out.csv", options={"delimiter": ";"}, engine="pandas")
```

### Loading a file, then tweaking it in Python before running

`load_config()` returns a real `PipelineConfig` (a Pydantic model), so you
can mutate fields — e.g. redirect runtime directories for a test run — before
passing it to `run_pipeline`:

```python
from ikietl import load_config, validate, run_pipeline, status, audit, clean

cfg = load_config("ikietl.yaml")
cfg.runtime.work_dir = "/tmp/my-run/work"
cfg.runtime.state_dir = "/tmp/my-run/state"

validate(cfg)                # raises PipelineError if schema/data checks fail
run_pipeline(cfg)            # full run, load included
print(status(cfg))           # {"timestamp": ..., "datasets": {...}}
print(audit(cfg))            # raw lineage.jsonl contents
clean(cfg)                   # tidy up work/tmp when you're done
```

`run_pipeline`/`validate`/`status`/`audit`/`clean` all accept **either** a
`PipelineConfig` object **or** a path string — pick whichever is convenient
at each call site; both forms go through the identical schema validation and
engine code the CLI uses, so behavior never diverges between "config file"
and "pure Python" usage.

---

## Reference: every extractor / transform op / validate rule / load mode

| Stage                      | Options                                                                              | Notes                                                                                                                                                                                    |
| -------------------------- | ------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Extract → file**         | `path`, `format`, `options` (incl. `engine`), `glob`                                 | Any `fsspec` location: local, `s3://`, `gs://`, `abfs://`, `http(s)://`, `sftp://`. `glob: true` concatenates every matched file. `options.engine` picks `pandas` (default) or `polars`. |
| **Extract → sql**          | `conn`, `query`, `params`                                                            | Any `SQLAlchemy`-supported database. `params` are bound as real query parameters (not string-interpolated), so this is not vulnerable to SQL injection from `${state.*}` values.         |
| **Formats (read + write)** | 17 pandas formats + 8 polars formats — see [I/O engines](#io-engines-pandas--polars) | Table-driven registry, one row per `(format, engine)`, shared by every extractor/loader. `format:` is optional and inferred from the file extension when omitted.                        |
| **Transform ops**          | `rename`, `select`, `drop`, `cast`, `filter`, `derive`, `join`, `sort`, `dedup`      | `filter`/`derive` run through a restricted evaluator — no arbitrary code exec. **`custom` is present but currently broken — see caveat above.**                                          |
| **Validate rules**         | `not_null`, `unique`, `range`, `row_count`                                           | In `strict` mode any failure aborts before load; in non-strict mode it's logged and the run continues.                                                                                   |
| **Load → file**            | `path`, `format`, `options` (incl. `engine`)                                         | Atomic write-then-rename. `{date}` in `path` is replaced with today's date. `options.engine` picks `pandas` (default) or `polars`.                                                       |
| **Load → sql**             | `conn`, `table`, `mode`, `transaction`                                               | `mode`: `append` or `replace` only — **no `upsert`**, and the `key:` field is currently unused.                                                                                          |

---

## CLI reference

| Command           | What it does                                                                              |
| ----------------- | ----------------------------------------------------------------------------------------- |
| `etl init [name]` | Scaffold a new `ikietl.yaml` plus `work/ logs/ state/ tmp/`                               |
| `etl validate`    | Config schema **and** data-contract checks — runs extract → transform → validate, no load |
| `etl dry-run`     | Same stages as `validate`; framed as "no load executed"                                   |
| `etl run`         | The full pipeline, including load                                                         |
| `etl status`      | Prints the contents of `state/last_run.json` (timestamp + per-dataset row counts)         |
| `etl audit`       | Prints `logs/audit/lineage.jsonl` — one JSON line per stage per dataset                   |
| `etl clean`       | Deletes `work/` and `tmp/`; keeps `state/`                                                |

All commands accept `--config`/`-c <path>`. Every one of them is a thin
wrapper over the Python API — see [Python API](#python-api) above for the
full picture, including running pipelines with no config file at all.

---

## Error-proofing, at a glance

| Feature                  | How                                                                                                            |
| ------------------------ | -------------------------------------------------------------------------------------------------------------- |
| Config schema validation | Pydantic model catches structural errors before `run`                                                          |
| Fail-fast / strict mode  | Any validation rule failure aborts the run (configurable)                                                      |
| Transactional SQL loads  | SQL loads wrapped in a transaction by default (`transaction: true`)                                            |
| Atomic file loads        | Written to `path.tmp` then renamed — never a half-written file                                                 |
| Format/engine mismatches | Clear `ValueError` naming the format, engine, and every format that engine supports — not a bare library error |
| State tracking           | `state/last_run.json` — timestamp of each run + row counts per dataset                                         |
| Audit / lineage          | Each extract/transform stage appends `{timestamp, stage, dataset_id, ...}` to `logs/audit/lineage.jsonl`       |
| Secrets                  | `${env.VAR}` interpolation — credentials never hardcoded in config                                             |
| Exit codes               | `0` success · `1` validation · `2` extract · `3` transform · `4` load · `5` config                             |

Sample `etl audit` output (actual field names — `timestamp`/`dataset_id`, not
`ts`/`dataset`):

```jsonl
{"timestamp": "2026-07-26T10:00:00+00:00", "stage": "extract", "dataset_id": "sales_csv", "rows": 1000, "type": "file"}
{"timestamp": "2026-07-26T10:00:01+00:00", "stage": "transform", "dataset_id": "clean_sales", "input": "sales_csv", "rows": 950}
```

Note: only `extract` and `transform` stages are logged on success today; a
failed non-strict `validate` rule is logged with `"failed": true`, and
successful validations and every `load` currently produce no lineage entry.

---

## Project layout

```
project/
├── ikietl.yaml           # your config (yaml/toml/json — pick one)
├── work/                 # intermediate/output artifacts written by `etl run`
├── state/                # last_run.json (timestamp + row counts)
├── logs/
│   └── audit/            # lineage.jsonl
├── tmp/                  # scratch, cleared by `etl clean`
└── examples/             # runnable ETL scenarios and per-example workspaces
    ├── basic-example/
    ├── database-etl/
    ├── cloud-storage-etl/
    └── ...
```

`etl init` scaffolds `work/`, `logs/`, `state/`, and `tmp/` for you alongside
a starter `ikietl.yaml`.

---

## Testing

```bash
pip install -e .
pip install pytest
pytest
```

`tests/test_pipeline_end_to_end.py` builds a temporary config + CSV on the
fly and runs it through the real pipeline (`run_pipeline`, the CLI's
`validate` command, and the Python API's `validate`/`status`), asserting on
row counts, output file contents, and config-relative path resolution.

`tests/test_examples.py` runs the shipped example scenarios end to end.

**Known gap:** the current test suite does not exercise `op: custom`,
`mode: upsert`, `runtime:`-omitted configs, `parallelism > 1`, or the
`engine: polars` I/O path — which is why some of these issues went
unnoticed. If you add tests for these, please contribute them back.

---

## License

MIT
