Metadata-Version: 2.4
Name: ikietl
Version: 0.1.0
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
Description-Content-Type: text/markdown

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

A single, table-driven ETL Swiss-army-knife. One optional-format config file
(YAML/TOML/JSON) declares every stage — **extract → transform → validate →
load** — against any file location or any database, and a small facade CLI
(`etl`) hides the pipeline DAG underneath.

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

- **Table-driven** — the pipeline is arrays of declarative stage-specs, not code
- **Single config** — no scattered `.env` files, no CLI flags carrying logic
- **Universal source** — any DB (`SQLAlchemy`) + any storage backend (`fsspec`) + any file format (registry)
- **Solid transformers** — declarative ops for the common 80%, real testable Python for the hard 20%
- **Error-proof** — schema-validated config, strict mode, transactions, checksums, clear exit codes
- **Examples-first** — a growing gallery of runnable ETL scenarios, each with its own structured workspace under the examples folder
- **Local backends** — optional Docker Compose stacks for PostgreSQL, MySQL, SQL Server, and MinIO so database and S3-style examples can be exercised locally

---

## Install

```bash
pip install ikietl
```

One package, no extras to remember — it ships with everything the tool
supports out of the box: cloud filesystems (`s3fs`, `gcsfs`, `adlfs`), common
database drivers (`psycopg2-binary`, `pymysql`), and Excel (`openpyxl`).

Editable/dev install: `pip install -e .`

---

## 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, framed as "no load executed"
etl run                             # execute the full pipeline, including load
etl status                          # show last run's state (last_run timestamp, watermarks)
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. Every command below works identically regardless of which format you
use — all three parse into the same Pydantic model.

### Try the example gallery

The repository ships with several runnable example scenarios under the examples
folder, each with its own generated workspace:

```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())"
```

You can also launch the local backends used by those scenarios:

```bash
docker compose -f examples/database-etl/docker-compose.yml up -d
docker compose -f examples/cloud-storage-etl/docker-compose.yml up -d
```

---

## Full feature tour

The config below is deliberately over-built — it isn't a "getting started"
example, it exercises **every** extractor, transform op, validation rule, and
load mode the codebase ships with, so you can see the whole surface area in
one place and delete what you don't need.

```yaml
meta:
  name: full-demo
  version: "1.0"
  description: "Every feature in one pipeline"

runtime:
  work_dir: ./work
  log_dir: ./logs
  state_dir: ./state
  dry_run: false
  strict: true # any failed validation rule aborts before load
  parallelism: 4

# ── EXTRACT ────────────────────────────────────────────────────────────────
# location (fsspec: local/S3/GCS/Azure/HTTP/SFTP) × database (SQLAlchemy)
# × format (registry) — composed independently, per source.
extract:
  - id: sales_csv
    type: file
    path: "s3://my-bucket/incoming/sales_*.csv" # fsspec handles the backend
    format: csv
    options: { delimiter: ",", header: true, encoding: "utf-8" }
    glob: true # read + concat every match

  - id: customers_db
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/db" # secrets never hardcoded
    query: "SELECT * FROM customers WHERE updated_at > :last_run"
    params: { last_run: "${state.last_run}" } # last run's watermark, auto-filled

  - id: events_json
    type: file
    path: "gs://analytics/events.jsonl"
    format: jsonl

# ── TRANSFORM ──────────────────────────────────────────────────────────────
# Declarative tier: rename, select, drop, cast, filter, derive, join, sort,
# dedup. filter/derive expressions run through a restricted evaluator
# (simpleeval) — never raw eval() — so a config file can't execute arbitrary code.
transform:
  - id: clean_sales
    input: sales_csv
    steps:
      - op: rename
        map: { "Order ID": order_id, "Amount": amount }
      - op: cast
        column: amount
        type: decimal
      - op: filter
        expr: "amount > 0"
      - op: derive
        column: year
        expr: "year(order_date)" # whitelisted function, not raw Python
      - op: select
        columns: [order_id, amount, year, customer_id, order_date]
      - op: join
        with: customers_db
        on_column: customer_id
        type: left
      - op: sort
        by: [order_date]
        ascending: true
      - op: dedup
        columns: [order_id]
      - op: custom # the hard 20% — real Python
        module: "transforms.sales"
        function: "normalize_currency"

# ── VALIDATE ───────────────────────────────────────────────────────────────
# All four rule types. In strict mode (default), any failure aborts before load.
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 }

# ── LOAD ───────────────────────────────────────────────────────────────────
# file (atomic write+rename) and sql (insert / append / replace / upsert,
# transactional) — every mode shown here for reference.
load:
  - id: warehouse_upsert
    input: clean_sales
    type: sql
    conn: "postgresql://user:${env.DB_PASS}@host/warehouse"
    table: fact_sales
    mode: upsert # insert | append | replace | upsert
    key: [order_id]
    transaction: true

  - id: archive_parquet
    input: clean_sales
    type: file
    path: "s3://my-bucket/archive/sales_{date}.parquet" # {date} auto-filled
    format: parquet

  - id: archive_csv
    input: clean_sales
    type: file
    path: "./work/clean_sales.csv"
    format: csv
    options: { delimiter: ";" }
```

Your custom transform, wired in above via `op: custom`:

```python
# transforms/sales.py
import pandas as pd

def normalize_currency(df: pd.DataFrame, **kwargs) -> pd.DataFrame:
    """Real, testable Python for logic too complex to express declaratively."""
    df["amount"] = df["amount"].round(2)
    return df
```

Run it:

```bash
etl validate -c full-demo.yaml    # catches schema typos AND data-contract failures
etl dry-run  -c full-demo.yaml    # extract -> transform -> validate, no load
etl run      -c full-demo.yaml    # the whole thing, load included
etl status   -c full-demo.yaml
etl audit    -c full-demo.yaml
```

---

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

| Stage                      | Options                                                                                   | Notes                                                                                                                                                                                                                                      |
| -------------------------- | ----------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **Extract → file**         | `path`, `format`, `options`, `glob`                                                       | Any `fsspec` location: local, `s3://`, `gs://`, `abfs://`, `http(s)://`, `sftp://`. `glob: true` concatenates every matched file.                                                                                                          |
| **Extract → sql**          | `conn`, `query`, `params`                                                                 | Any `SQLAlchemy`-supported database. `params` can reference `${state.*}` watermarks.                                                                                                                                                       |
| **Formats (read + write)** | `csv`, `tsv`, `json`, `jsonl`, `parquet`, `xlsx`, `orc`                                   | One registry entry per format, shared by every extractor/loader. `format` is explicit and overridable — bad/ambiguous paths fail at `etl validate`, not mid-run.                                                                           |
| **Transform ops**          | `rename`, `select`, `drop`, `cast`, `filter`, `derive`, `join`, `sort`, `dedup`, `custom` | `filter`/`derive` expressions support `year()`, `month()`, `upper()`, `lower()`, `round()`, `abs()` — whitelisted, no arbitrary code exec. `join` uses `on_column` (not `on` — a bare `on:` is parsed as the YAML 1.1 boolean key `True`). |
| **Validate rules**         | `not_null`, `unique`, `range`, `row_count`                                                | In `strict` mode, any failure raises and aborts before load; in non-strict mode it's logged to the audit trail and the run continues.                                                                                                      |
| **Load → file**            | `path`, `format`, `options`                                                               | Atomic: written to `path.tmp` then renamed — never a half-written file. `{date}` in `path` is replaced with today's date.                                                                                                                  |
| **Load → sql**             | `conn`, `table`, `mode`, `key`, `transaction`                                             | `mode`: `insert`/`append` (plain insert), `replace` (drop+recreate), `upsert` (key-based, transactional, one dialect's `ON CONFLICT` shown — swap for `MERGE`/`ON DUPLICATE KEY` per dialect as needed).                                   |

---

## CLI reference

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

All commands accept `--config`/`-c <path>`.

---

## 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                                         |
| Idempotent loads         | `insert` / `replace` / `upsert` / `append`, key-based                              |
| Transactional loads      | SQL loads wrapped in a transaction; file loads use atomic write+rename             |
| State tracking           | `state/last_run.json` — timestamps, watermarks                                     |
| Audit / lineage          | Every stage logs `{stage, dataset, rows}` to `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:

```jsonl
{"ts": "2026-07-26T10:00:00+00:00", "stage": "extract", "dataset": "sales_csv", "rows": 1000}
{"ts": "2026-07-26T10:00:01+00:00", "stage": "transform", "dataset": "clean_sales", "rows": 950}
{"ts": "2026-07-26T10:00:01+00:00", "stage": "validate", "dataset": "sales_quality", "status": "passed"}
{"ts": "2026-07-26T10:00:02+00:00", "stage": "load", "dataset": "warehouse_upsert", "rows": 950}
```

---

## Project layout

```
project/
├── ikietl.yaml          # your config (yaml/toml/json — pick one)
├── transforms/          # your custom Python transform modules
│   └── sales.py
├── work/                # intermediate datasets
├── state/                # watermarks, checksums, last_run.json
├── logs/                 # structured + human logs
├── audit/                # lineage records
├── tmp/                 # scratch, cleared by `etl clean`
└── examples/            # runnable ETL scenarios and per-example workspaces
    ├── basic-example/
    ├── database-etl/
    ├── cloud-storage-etl/
    └── ...
```

---

## Testing

```bash
pytest
```

`tests/test_pipeline_end_to_end.py` runs the shipped example
(`examples/ikietl.yaml` + `examples/sample_sales.csv`) through the real
pipeline and asserts on the row count that survives the `filter` step.

---

## License

MIT
