lotos 0.1.0

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

🪷 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
• Secret resolution — reference environment variables or .env values with ${SECRET:KEY} syntax
• Incremental loading — watermark-based extraction for processing only new/changed rows
• Structured logging — powered by structlog with JSON or console output
• CLI interface — run, validate, inspect, and scaffold pipelines from the terminal

---

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 ".[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
  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

3. Run it

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

---

CLI Commands

Command	Description
lotos run <file>	Run a pipeline
lotos validate <file>	Validate YAML without running
lotos inspect <file>	Show pipeline details
lotos init <name>	Generate a pipeline template
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

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

---

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

---

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/                   # Pipeline engine, models, registry
├── sinks/                  # Destination sinks (file, ADLS Gen2)
├── transforms/             # Transform plugins
└── monitoring/             # Monitoring (placeholder)
pipelines/                  # Pipeline YAML definitions
tests/                      # Test suite

---

Development

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

---

License

Artefact