ikietl 0.1.0

Table-driven ETL Swiss-knife with CLI and Python API

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

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

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:

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:

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.

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:

# 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:

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:

{"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

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