Metadata-Version: 2.4
Name: lotos
Version: 1.0.0
Summary: Self-hosted data ingestion framework — extract, transform, and load data from anywhere.
Author: yob
License: ARTEFACT
Requires-Python: >=3.11
Requires-Dist: httpx>=0.27
Requires-Dist: polars>=1.0.0
Requires-Dist: pydantic-settings>=2.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pymysql
Requires-Dist: pyyaml>=6.0
Requires-Dist: rich>=13.0
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: structlog>=24.0
Requires-Dist: typer[all]>=0.12
Provides-Extra: all
Requires-Dist: azure-identity>=1.0; extra == 'all'
Requires-Dist: azure-storage-blob>=12.0; extra == 'all'
Requires-Dist: azure-storage-file-datalake>=12.0; extra == 'all'
Requires-Dist: psycopg2-binary>=2.9; extra == 'all'
Requires-Dist: pymysql>=1.1; extra == 'all'
Requires-Dist: pyodbc>=5.0; extra == 'all'
Provides-Extra: azure
Requires-Dist: azure-identity>=1.0; extra == 'azure'
Requires-Dist: azure-storage-blob>=12.0; extra == 'azure'
Requires-Dist: azure-storage-file-datalake>=12.0; extra == 'azure'
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: monitoring
Requires-Dist: prometheus-client>=0.20; extra == 'monitoring'
Provides-Extra: mssql
Requires-Dist: pyodbc>=5.0; extra == 'mssql'
Provides-Extra: mysql
Requires-Dist: pymysql>=1.1; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9; extra == 'postgres'
Description-Content-Type: text/markdown

# 🪷 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
![Architecture Diagram](archi.png)


---

## Installation

```bash
pip install -e .
```

With optional dependencies:

```bash
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

```bash
lotos init my_pipeline --source sql
```

### 2. Edit the YAML

```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

```bash
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.

```yaml
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.

```yaml
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

```yaml
pipeline:
  name: load_orders
  depends_on:
    - load_customers
    - load_products
```

### Running the DAG

```bash
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:

```yaml
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

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

### Run History

```bash
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:

```bash
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:

```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:

```bash
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:

```yaml
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

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

---

## License

Artefact
