ETLai

A local, AI-assisted data transformation engine built on Dagster. Install it, scaffold a project, and let Claude Code create new pipelines — from CSV transformations to API ingestion.

Install

pip install ETLai

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

Quick start

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

# Option A: Create a pipeline with the 5-agent system
etlai create "join sales data with product catalog and compute revenue"

# Option B: Manual pipeline creation (see ARCHITECTURE.md)
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

How it works

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

• Atoms — reusable, single-unit-of-work transformations (domain-agnostic)
• Triggers — what causes the pipeline to run (file sensor, cron, or both)
• Config — business logic (column names, thresholds, expressions) in config.json

Trigger fires → load files → read config.json → execute steps
                               ├→ processed/ + output/     (success)
                               └→ rejected/ + *.error.txt  (failure)

Pipeline types

Single atom (file-based)

Drop files into inbox/. Sensor detects stable files and triggers the pipeline.

name: vlookup_rollnumber
atom: vlookup
min_files: 2
inputs_map:
  - param: left_file
  - param: right_file

Its parameters live in pipelines/vlookup_rollnumber/config.json.

API-based (data ingestion)

Fetch data from REST APIs on a schedule. No inbox files needed.

name: fetch_hr_data
atom: api_fetch
min_files: 0
env_file: ~/.etlai/secrets.env
requires_env: [HR_API_TOKEN]
trigger:
  rules:
    - type: schedule
      cron: "0 8 * * *"

Composite (multi-step chains)

Chain multiple atoms with reference file injection and branching.

name: sales_reconciliation
min_files: 1
inputs:
  - name: sales_data
    role: transient
    description: "Weekly sales CSV"
steps:
  - atom: vlookup                # step 0: join sales + catalog
    inputs_map:
      - param: left_file
      - param: right_file
        source: reference        # permanent lookup matched by pattern
        pattern: "catalog.csv"
  - atom: computed_column        # step 1: compute revenue
    inputs_map:
      - param: input_file
        source: prev_output
  - atom: group_aggregate        # step 2: aggregate by category
    input_from: 1                # reads step 1, not step 2 (branching)
    inputs_map:
      - param: input_file
        source: prev_output
  - atom: rename_columns         # final step: rehydrate column names
    inputs_map:
      - param: input_file
        source: prev_output
trigger:
  rules:
    - type: inbox_files
      min_files: 1

Shipped atoms (10)

Creating pipelines with AI

The 5-agent system (etlai create) orchestrates pipeline creation through 7 phases with deterministic gate validators between each:

The firewall physically hides business_mapping.json from Atom Smith, ensuring atoms remain domain-agnostic by construction.

See ARCHITECTURE.md for full details on the agent system.

Key concepts

reference (permanent lookup files)

Permanent lookup files in reference/ bind to a step param via source: reference:

steps:
  - atom: vlookup
    inputs_map:
      - param: left_file
      - param: right_file
        source: reference
        pattern: "catalog.csv"

input_from (non-linear step routing)

Steps read the previous step's output by default. Use input_from to branch:

steps:
  - atom: computed_column     # step 0
  - atom: rename_columns      # step 1 (reads step 0)
  - atom: group_aggregate     # step 2 reads step 0, NOT step 1
    input_from: 0

config.json

Every step reads its step_N key (including step_0):

{
  "step_0": {"left_column": "sku", "right_column": "sku"},
  "step_1": {"expression": "price * qty", "output_column": "revenue"},
  "step_2": {"mapping": {"revenue": "Total Revenue"}}
}

Credentials

API credentials live in env files outside the project (e.g. ~/.etlai/secrets.env). etlai sync validates required vars are present without printing values.

Resolution order

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

User atoms take precedence over shipped ones.

Development

git clone <repo>
pip install -e ".[dev]"
pytest --cov=etlai

See CONTRIBUTING.md for workflow, ARCHITECTURE.md for internals, and TESTS.md for testing guide.