ETLai 0.2.2

Local, folder-driven Dagster CSV transformation engine with hot-folder sensors, AI-assisted pipeline creation, and Tkinter configuration UI.

ETLai

A local, AI-assisted CSV transformation engine built on Dagster. Install it, scaffold a project, and let Claude Code create new pipelines for you.

Install

pip install ETLai

Requires Python 3.10+ and Tkinter (ships with most Python installations).

Quick start

etlai init ~/my-etl
cd ~/my-etl
etlai sync
etlai run

This starts a Dagster dev server at http://localhost:3000. Enable sensors in the UI, then drop CSV files into pipelines/<name>/inbox/.

Commands

Command	Purpose
etlai init <dir>	Scaffold a new project with example pipelines
etlai sync	Validate manifests, create folders, prompt for path: ask
etlai run	Start the Dagster dev server
etlai list	Show all registered pipelines

How it works

Each pipeline is defined by a manifest (pipelines/<name>/manifest.yaml) that wires together:

• Atom — a reusable, domain-agnostic transformation (execute(params_json) -> result_json)
• Form — a first-run Tkinter UI that collects user config (or passthrough for no-UI pipelines)
• Sensor — watches the inbox folder for stable files

inbox/ → sensor (stability check) → staging/ → job runs
                                      ├→ processed/ + output/     (success)
                                      └→ rejected/ + *.error.txt  (failure)

Config is saved to config.json after first run. Subsequent runs are fully automated. Delete config.json to force reconfiguration.

Shipped atoms

Atom	Input	Description
vlookup	2 CSV	Left join on specified columns with dtype validation
groupby	1 CSV	Group by column with count, sorted descending
mock_generate	1+ CSV	Generate synthetic data from file headers using Faker

Shipped forms

Form	Description
vlookup_column_picker	Join column + output column multi-select with dtype validation
groupby_picker	Single column selection
passthrough	No UI — reads config.json and passes it to atom

Example manifest

name: vlookup_rollnumber
atom: vlookup
form: vlookup_column_picker
min_files: 2
path: ask                     # prompts for folder location during etlai sync

Composite pipelines

Chain multiple atoms in sequence:

name: vlookup_then_groupby
min_files: 2
steps:
  - atom: vlookup
    form: vlookup_column_picker
  - atom: groupby
    form: groupby_picker

Output of each step becomes input to the next.

Custom data paths

Each pipeline can store its data anywhere. Set path in the manifest:

• path: ask — etlai sync opens a folder picker dialog
• path: /absolute/path — uses that path directly
• Omitted — defaults to pipelines/<name>/ inside the project

Adding new pipelines with Claude Code

Open your project in Claude Code. The scaffolded CLAUDE.md teaches it how to:

1. Create atoms in atoms/ (or reuse shipped ones)
2. Create forms in forms/ (or use passthrough with a pre-written config.json)
3. Write a manifest.yaml in pipelines/<name>/
4. Run etlai sync to validate and create folders

For no-UI pipelines, Claude Code writes config.json directly with the business logic (e.g. {"group_column": "religion"}).

Resolution order

• Atoms: ./atoms/<name>.py → etlai.atoms.<name> (package)
• Forms: ./forms/<name>.py → etlai.forms.<name> (package)

User files take precedence over shipped ones.

Project structure after etlai init

my-etl/
  etlai.yaml              ← pipelines_root config
  dagster.yaml            ← auto-generated storage config
  definitions.py          ← 3-line auto-loader
  CLAUDE.md               ← instructions for Claude Code
  atoms/                  ← custom atoms (AI-generated)
  forms/                  ← custom forms (AI-generated)
  pipelines/
    <name>/
      manifest.yaml       ← wiring: atom + form + min_files + path
      config.json         ← saved config (always here, even if path points elsewhere)
      inbox/              ← drop files here (or at custom path location)
      staging/            ← in-flight processing
      processed/          ← successfully processed source files
      rejected/           ← failed files + .error.txt
      output/             ← transformation results

Development

git clone <repo>
pip install -e .
etlai init /tmp/test-project
cd /tmp/test-project
etlai run

No automated tests or linter configured yet.