loafer-etl 0.3.1

AI-assisted ETL and ELT pipelines from the command line

Loafer

A CLI-first, declarative ETL/ELT pipeline tool driven by YAML.

Overview

Modern data pipelines often become bogged down by repetitive boilerplate and fragile scripts. Loafer solves this by treating data movement and transformation as configuration.

Loafer is a CLI-first tool that allows you to extract data from databases, files, and APIs, apply powerful transformations (both custom and AI-generated), and load the results into target systems — all configured within a single, highly readable YAML file.

Key Idea: Describe your pipeline in YAML, run it from the CLI, and let Loafer handle the execution.

Features

• Declarative Configuration: Define your inputs, outputs, and processing logic in simple YAML. No complex workflow orchestrators required.
• ETL & ELT Support: Choose between Extract-Transform-Load (in-memory/streaming Python transformations) or Extract-Load-Transform (in-database SQL executions).
• AI-Powered Transformations: Optionally use natural language to auto-generate complex Python transforms or target-database SQL using modern LLMs (Gemini, OpenAI, Claude).
• Streaming & Batch Processing: Automatically switches to memory-efficient streaming for large datasets based on configurable thresholds.
• Data Quality Guards: Built-in validation steps to catch malformed data or high null-rates before they hit your target.
• Developer-Friendly & Extensible: Use custom Python files for advanced transformations when declarative logic isn't enough.

Installation

Via pip (PyPI)

The easiest way to install Loafer is via pip:

pip install loafer-etl

Via Docker (GHCR)

Loafer is officially published to the GitHub Container Registry. Available tags include specific versions (e.g. 0.2.0, 0.2) and latest.

docker pull ghcr.io/lupppig/loafer:latest

To run Loafer using Docker, mount your current working directory so Loafer can access your configuration and local files:

docker run --rm -v $(pwd):/workspace -w /workspace ghcr.io/lupppig/loafer:latest run pipeline.yaml

Warning
Do not mount your working directory to /app (e.g., -v $(pwd):/app). The Loafer image uses /app internally for its application core and virtual environment. Overwriting this directory will break the container. Always use /workspace or another path.

From Source

To contribute or use the latest unreleased features:

git clone https://github.com/lupppig/loafer.git
cd loafer
pip install -e .

Quick Start

Create a single YAML file describing your pipeline. This example extracts orders from PostgreSQL, normalizes the data via AI, and saves it to a clean CSV.

pipeline.yaml

name: Daily Orders Pipeline
mode: etl

source:
  url: ${DATABASE_URL}
  query: "SELECT * FROM orders WHERE created_at >= NOW() - INTERVAL '1 day'"

target:
  path: ./output/clean_orders.csv
  write_mode: overwrite

transform: >
  Drop cancelled orders, normalize currency to USD, and 
  combine first_name and last_name into full_name.

llm:
  provider: gemini
  model: gemini-2.5-flash
  api_key: ${GEMINI_API_KEY}

Note: Loafer auto-detects source and target types from URL schemes and file extensions — no need to write type: postgres or type: csv. See Auto-Detection below.

Run the pipeline from your terminal:

export DATABASE_URL="postgresql://user:pass@localhost/db"
export GEMINI_API_KEY="your-api-key"

loafer run pipeline.yaml

Expected Outcome: Loafer will connect to the Postgres database, extract the past day's orders, execute the transformation to clean the data, and successfully write the normalized output to ./output/clean_orders.csv.

How It Works

Loafer pipelines consist of three main stages, executed as a directed graph:

1. Extract: Loafer connects to your source (e.g., a SQL database, CSV, or REST API) and pulls the data. Large datasets are automatically chunked and streamed to prevent memory exhaustion.
2. Transform: The data is manipulated according to your YAML instructions. This can be AI-generated Python code running in a safe isolated context, a custom Python script you provide, or skipped entirely for simple EL (Extract-Load) operations.
3. Load: The transformed data is pushed to your designated target connector (e.g. CSV, Snowflake, Postgres) adhering to your specified write mode (append, overwrite).

In ELT mode, the steps differ slightly: data is first loaded raw into the target database, and transformations are executed as native SQL queries against the target engine.

CLI Usage

Loafer provides a streamlined CLI tailored for day-to-day data engineering.

Command Syntax:

loafer <command> [options]

Common Commands:

• loafer run <config.yaml>: Execute a pipeline defined in the given YAML file.
  • --dry-run: Extracts and transforms data without writing to the target.
  • --verbose: Prints detailed execution logs and agent outputs.
• loafer validate <config.yaml>: Check a configuration file for syntax and connection errors without running the pipeline.
• loafer connectors: List all available source and target connectors.

Project Structure

loafer/
├── cli.py             # CLI entrypoints and command routing
├── config.py          # Configuration parsing and validation
├── runner.py          # Pipeline execution logic and LangGraph orchestration
├── connectors/        # Integrations (sources and targets)
│   ├── sources/          # Postgres, CSV, REST API, Excel, etc.
│   └── targets/          # Postgres, CSV, Snowflake, etc.
├── transform/         # Transformation engines (AI, custom, SQL)
├── graph/             # LangGraph state management and pipeline DAGs
├── llm/               # LLM provider integrations (Gemini, Claude, OpenAI)
└── agents/            # Individual workflow nodes (extract, transform, load)

Auto-Detection

Loafer automatically infers connector types so you can write minimal YAML.

Sources — detected from url scheme or file path extension:

Signal	Detected Type
postgresql:// / postgres://	postgres
mysql:// / mysql+pymysql://	mysql
mongodb:// / mongodb+srv://	mongo
http:// / https://	rest_api
.csv	csv
.xlsx / .xls	excel
.pdf	pdf
.db / .sqlite / .sqlite3	sqlite

Targets — detected from url scheme or file path extension:

Signal	Detected Type
postgresql:// / postgres://	postgres
mongodb:// / mongodb+srv://	mongo
.csv	csv
.json / .jsonl	json

Transforms — detected from which fields are present:

Signal	Detected Type
instruction field present	ai
path field present	custom
query field present	sql

If Loafer can't detect the type (e.g., a file with no extension), it will ask you to add an explicit type: field. You can always specify type: manually to override auto-detection.

Incremental Loading

For SQL sources (postgres, mysql, sqlite) and rest_api, Loafer can extract only rows newer than the previous run instead of re-pulling the full dataset — ideal for scheduled pipelines.

source:
  url: ${DATABASE_URL}
  query: SELECT * FROM orders

incremental:
  column: updated_at      # cursor column (or REST response field)
  initial: "1970-01-01"   # watermark used on the first run

Loafer remembers the highest updated_at it has seen in <config>.loafer-state.json (next to your config) and only advances the watermark after a fully successful load. Use loafer run pipeline.yaml --full-refresh to ignore the saved cursor and re-pull everything.

Upsert (Idempotent Loads)

PostgreSQL and MongoDB targets support write_mode: upsert so re-running a pipeline updates existing rows instead of duplicating them — the natural companion to incremental loading.

target:
  url: ${TARGET_DB_URL}
  table: orders_synced
  write_mode: upsert
  key: order_id          # single column, or [tenant_id, order_id] for a composite key

Postgres uses INSERT ... ON CONFLICT (key) DO UPDATE (auto-creating a unique index on the key if needed); MongoDB uses a bulk ReplaceOne(..., upsert=True). key is required when write_mode is upsert.

Sandboxed Transforms

AI-generated and custom Python transforms run in an isolated, resource-limited subprocess, not in the main process. On Linux/macOS the worker is bounded by CPU and memory rlimits and a hard timeout, so a runaway or malicious transform is killed with a clear error instead of hanging the pipeline or reaching your credentials. Tune the limits with an optional sandbox block:

sandbox:
  timeout: 60        # seconds before the worker is killed
  max_memory_mb: 512 # address-space cap for the worker

On Windows the sandbox degrades to an in-process best-effort timeout (resource limits aren't enforced) — run on Linux, including the Docker image, for full isolation.

Configuration (YAML)

Loafer pipelines are driven by a single YAML configuration file. The type field is optional — Loafer will auto-detect it when possible.

# Pipeline metadata
name: User Sync Pipeline
mode: etl  # Supports 'etl' or 'elt'

# Extract — type auto-detected as rest_api from https:// URL
source:
  url: "https://api.example.com/users"
  method: GET

# Load — type auto-detected as postgres from postgresql:// URL
target:
  url: ${TARGET_DB_URL}
  table: users_dim

# Transform — type auto-detected as custom from path field
transform:
  path: ./transforms/clean_users.py

# Optional: LLM Configuration (required for AI transforms or ELT mode)
llm:
  provider: gemini
  model: gemini-2.5-flash
  api_key: ${GEMINI_API_KEY}

# Performance & Validation
chunk_size: 1000
streaming_threshold: 10000

validation:
  strict: true
  max_null_rate: 0.1

LLM Setup

To use the AI transformations (via type: ai) or dynamic SQL generation in ELT mode, you must specify an LLM provider in your configuration. We recommend setting API keys via environment variables for security.

Loafer supports four providers out-of-the-box:

1. Gemini (Default)

llm:
  provider: gemini
  model: gemini-2.5-flash
  api_key: ${GEMINI_API_KEY}

2. OpenAI

llm:
  provider: openai
  model: gpt-4o
  api_key: ${OPENAI_API_KEY}

3. Claude (Anthropic)

llm:
  provider: claude
  model: claude-3-5-sonnet-20241022
  api_key: ${ANTHROPIC_API_KEY}

4. Qwen (Alibaba)

llm:
  provider: qwen
  model: qwen-max
  api_key: ${QWEN_API_KEY}

Development

Setting up Loafer for local development requires uv (or standard pip / hatch).

1. Clone and Install dependencies:

   git clone https://github.com/yourusername/loafer.git
   cd loafer
   # If using uv for fast dependencies:
   uv sync
   # Or using standard pip:
   pip install -e ".[dev]"
   

2. Run Tests: Loafer uses pytest for unit and integration testing.

   pytest
   

3. Code Style: This project uses standard Python linting and formatting tools. We recommend running ruff prior to committing:

   ruff check .
   ruff format .
   

Contributing

Contributions are highly encouraged! Whether it’s a new data connector, a bug fix, or an improvement to the documentation, we'd love your help.

1. Fork the repository.
2. Create a new branch for your feature (git checkout -b feature/amazing-connector).
3. Commit your changes with clear messages.
4. Open a Pull Request against the main branch.

Keep things simple and ensure any new features include appropriate tests.

Links

• GitHub Repository: https://github.com/lupppig/loafer
• PyPI Package: https://pypi.org/project/loafer-etl/

License

This project is licensed under the MIT License. See the LICENSE file for details.