🪷 Lotos

Self-hosted data ingestion framework — extract, transform, and load data from anywhere.

Lotos is a lightweight, YAML-driven ETL framework built with Python. Define pipelines as configuration, not code.

---

Features

• YAML-based pipelines — declarative, version-controlled, easy to read
• Pluggable architecture — connectors, transforms, and sinks are auto-discovered via a registry
• Source connectors — SQL databases, REST APIs, local/remote files (CSV, JSON, Parquet)
• Transform chain — select, rename, cast, filter, deduplicate, flatten, SQL transforms, computed columns, custom expressions, and schema validation
• Destination sinks — local files, Azure Blob Storage, Azure Data Lake Storage Gen2
• Full & incremental loading — watermark-based extraction with SQLite-persisted state
• DAG orchestration — dependency-aware parallel execution of pipeline graphs with retries and backoff
• Run history — every execution is tracked (status, row counts, duration, errors)
• State management — inspect, reset, and clear watermarks and history via CLI
• Secret resolution — reference environment variables or .env values with ${SECRET:KEY} syntax
• Structured logging — powered by structlog with JSON or console output
• Monitoring — Prometheus metrics endpoint + pre-built Grafana dashboard
• Data quality checks — row count, null %, freshness, uniqueness, value range, regex match
• CLI interface — run, validate, inspect, orchestrate, monitor, and scaffold pipelines from the terminal

---

Architecture

---

Installation

pip install -e .

With optional dependencies:

pip install -e ".[azure]"       # Azure Blob / ADLS Gen2
pip install -e ".[postgres]"    # PostgreSQL
pip install -e ".[mysql]"       # MySQL
pip install -e ".[monitoring]"  # Prometheus metrics + Grafana
pip install -e ".[all]"         # Everything
pip install -e ".[dev]"         # Dev tools (pytest, ruff)

Requires Python 3.11+

---

Quick Start

1. Generate a pipeline template

lotos init my_pipeline --source sql

2. Edit the YAML

pipeline:
  name: my_pipeline
  description: "Extract customers and clean data"
  version: "1.0"

source:
  connector: sql
  load_mode: incremental        # full (default) or incremental
  config:
    connection_string: "mysql+pymysql://user:pass@localhost:3306/mydb"
    query: "SELECT * FROM customers"

transforms:
  - type: select_columns
    config:
      columns: [id, name, email, created_at]

  - type: deduplicate
    config:
      subset: [id]
      keep: last

sink:
  connector: file
  config:
    path: "output/customers.parquet"
    format: parquet
  write_mode: overwrite

watermark:
  field: updated_at
  initial_value: "2020-01-01"

3. Run it

lotos run pipelines/my_pipeline.yaml --log-level INFO

---

Load Modes

Lotos supports two extraction strategies, configured per-pipeline via source.load_mode:

Full Load (full)

Extracts all rows from the source on every run. The watermark is ignored for filtering but still tracked for informational purposes.

source:
  connector: sql
  load_mode: full
  config:
    query: "SELECT * FROM orders"

Incremental Load (incremental)

Extracts only new/changed rows since the last run, using a watermark column. The watermark value is persisted in a local SQLite state store (.lotos/state.db) and automatically updated after each successful run.

source:
  connector: sql
  load_mode: incremental
  config:
    connection_string: "${SECRET:DB_CONN}"
    query: "SELECT * FROM orders WHERE updated_at > :watermark"

watermark:
  field: updated_at
  initial_value: "2024-01-01"

Watermark resolution order:

1. Explicit CLI override (--watermark)
2. Full mode → always None (no filter)
3. Incremental → persisted value from SQLite state store
4. Incremental, no persisted value → watermark.initial_value from YAML

---

Pipeline Orchestration

Orchestrate multiple pipelines as a directed acyclic graph (DAG). Independent pipelines run in parallel; dependent pipelines wait for their upstream dependencies.

Declaring dependencies

pipeline:
  name: load_orders
  depends_on:
    - load_customers
    - load_products

Running the DAG

lotos orchestrate pipelines/ --workers 4 --log-level INFO

Lotos will:

1. Discover all .yaml / .yml files in the directory
2. Build a dependency graph from depends_on declarations
3. Topologically sort into execution layers (Kahn's algorithm)
4. Execute each layer in parallel with a thread pool
5. Skip downstream pipelines if an upstream dependency fails (marked cancelled)
6. Retry failed pipelines with exponential backoff (2^attempt seconds, capped at 120s)
7. Track every run in the state store

Example DAG

pipelines/
├── 01_customers.yaml          # depends_on: []
├── 02_products.yaml           # depends_on: []
├── 03_orders.yaml             # depends_on: [customers, products]
└── 04_analytics.yaml          # depends_on: [orders]

Execution layers:

Layer 1: customers, products    (parallel)
Layer 2: orders                 (waits for layer 1)
Layer 3: analytics              (waits for layer 2)

Retry & schedule config

Configure retries and timeouts per pipeline:

schedule:
  max_retries: 3                # 1 initial + 3 retries = 4 attempts
  retry_backoff: exponential    # 2s, 4s, 8s, …, capped at 120s
  timeout_seconds: 3600         # per-pipeline timeout warning

---

State Management

All pipeline state is stored in .lotos/state.db (SQLite with WAL mode).

Watermarks

lotos state show                       # List all watermarks
lotos state reset my_pipeline          # Clear watermark (forces full reload)

Run History

lotos history                          # Show all runs (last 20)
lotos history my_pipeline              # Filter by pipeline
lotos history --status failed -n 50    # Filter by status, show 50
lotos state clear-history              # Delete all history
lotos state clear-history my_pipeline  # Delete history for one pipeline

---

CLI Commands

Command	Description
lotos run <file>	Run a single pipeline
lotos orchestrate <dir>	Run all pipelines as a DAG
lotos validate <file>	Validate YAML without running
lotos inspect <file>	Show pipeline details
lotos init <name>	Generate a pipeline template
lotos monitor	Start Prometheus metrics server
lotos quality <file>	Run quality checks only (no sink)
lotos list connectors	List available source connectors
lotos list transforms	List available transforms
lotos list sinks	List available destination sinks
lotos list all	List everything
lotos history [pipeline]	Show run history
lotos state show	Show all stored watermarks
lotos state reset <name>	Clear watermark for a pipeline
lotos state clear-history	Delete run history

Run options

--output, -o       Save result to file (csv/json/parquet)
--format, -f       Output format (default: parquet)
--dry-run          Extract + transform only, skip sink
--log-level, -l    DEBUG, INFO, WARNING, ERROR
--log-format       console or json

Orchestrate options

--workers, -w      Max parallel pipelines (default: 4)
--dry-run          Extract + transform only
--log-level, -l    Log level
--log-format       Log format

Monitor options

--port, -p         Prometheus HTTP port (default: 9090)
--addr             Bind address (default: 0.0.0.0)
--backfill         Backfill metrics from state store history

---

Monitoring & Data Quality

Prometheus metrics

Install with pip install -e ".[monitoring]", then:

lotos monitor --port 9090 --backfill   # start metrics server

Exposed metrics at http://localhost:9090/metrics:

Metric	Type	Labels
lotos_pipeline_runs_total	Counter	pipeline, status
lotos_pipeline_duration_seconds	Histogram	pipeline
lotos_pipeline_rows_extracted_total	Counter	pipeline
lotos_pipeline_rows_loaded_total	Counter	pipeline
lotos_pipeline_last_success_timestamp	Gauge	pipeline
lotos_pipeline_errors_total	Counter	pipeline
lotos_quality_checks_total	Counter	pipeline, check, status

Grafana dashboard

Import the pre-built dashboard from lotos/monitoring/grafana_dashboard.json into Grafana.
It includes 14 panels covering pipeline runs, duration percentiles, data volume, quality checks, and error trends.

Data quality checks

Add a quality_checks section to any pipeline YAML:

quality_checks:
  - type: row_count
    min: 100
    action: block        # block | warn | alert

  - type: null_percentage
    column: email
    max_percentage: 5.0
    action: warn

  - type: freshness
    column: updated_at
    max_age_hours: 24
    action: alert

  - type: uniqueness
    columns: [id]
    action: block

  - type: value_range
    column: age
    min: 0
    max: 150
    action: warn

  - type: regex_match
    column: email
    pattern: "^[\\w.+-]+@[\\w-]+\\.[a-zA-Z]{2,}$"
    action: alert

Run checks without loading:

lotos quality pipelines/my_pipeline.yaml

---

Connectors

Name	Type	Description
sql	Source	Any SQLAlchemy-compatible database
rest_api	Source	REST APIs with pagination & auth
file	Source	CSV, JSON, Parquet (local or Azure Blob)
file	Sink	Write to local files or Azure Blob
adls_gen2	Sink	Azure Data Lake Storage Gen2 / Blob Storage

ADLS Gen2 Sink

The adls_gen2 sink supports both Azure storage modes via the hierarchical_namespace option:

sink:
  connector: adls_gen2
  config:
    account_name: "mystorageaccount"
    container_name: "mycontainer"
    directory_path: "data/output"
    file_name: "results"
    format: parquet
    hierarchical_namespace: false  # false = Blob API, true = Data Lake (DFS) API
    account_key: "${SECRET:ADLS_KEY}"
  write_mode: overwrite

Authentication methods (in priority order):

1. account_key — storage account key
2. connection_string — full connection string
3. sas_token — shared access signature
4. DefaultAzureCredential — automatic (managed identity, env vars, CLI login)

---

Transforms

Name	Description
select_columns	Keep only specified columns
rename_columns	Rename columns via a mapping
cast_types	Cast column data types
filter_rows	Filter rows by conditions
deduplicate	Remove duplicate rows
flatten	Flatten nested structs/lists
sql	Run SQL queries (Polars SQL — no DB needed)
computed	Add computed columns with expressions
custom	Apply a custom Python expression
validate_schema	Validate column types and constraints

---

Project Structure

lotos/
├── cli.py                  # CLI (Typer)
├── config/                 # Settings, logging, secrets
├── connectors/             # Source connectors (SQL, REST, file)
├── core/
│   ├── models.py           # Pydantic models (YAML schema)
│   ├── pipeline.py         # Pipeline execution engine
│   ├── registry.py         # Plugin auto-discovery
│   ├── state.py            # SQLite state store
│   └── exceptions.py       # Custom exceptions
├── orchestration/
│   └── base.py             # DAG orchestrator
├── sinks/                  # Destination sinks (file, ADLS Gen2)
├── transforms/             # Transform plugins
└── monitoring/             # Monitoring & quality
    ├── base.py             # Monitor interfaces (Log / Prometheus)
    ├── metrics.py          # Prometheus metrics collector & server
    ├── quality.py          # Data quality checker (6 check types)
    └── grafana_dashboard.json  # Pre-built Grafana dashboard
pipelines/                  # Pipeline YAML definitions
tests/                      # Test suite

---

Development

pip install -e ".[dev]"
pytest
ruff check .

---

License

Artefact