Metadata-Version: 2.4
Name: etlpy-tools
Version: 1.0.0
Summary: A composable, async, Polars-native ETL toolkit
Author-email: Vladimir Novotorzhin <include.dos.h@yandex.ru>
License-Expression: MIT
Project-URL: Homepage, https://github.com/Sque-ak/etlpy
Project-URL: Repository, https://github.com/Sque-ak/etlpy
Project-URL: Issues, https://github.com/Sque-ak/etlpy/issues
Keywords: etl,polars,clickhouse,airflow,data-pipeline
Classifier: Development Status :: 4 - Beta
Classifier: Programming Language :: Python :: 3.13
Classifier: Intended Audience :: Developers
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.13
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: polars>=1.41.2
Requires-Dist: polars-hash>=0.6.0
Provides-Extra: extractor
Requires-Dist: httpx>=0.28.1; extra == "extractor"
Requires-Dist: etlpy-tools[loader]; extra == "extractor"
Provides-Extra: transformer
Requires-Dist: polars>=1.41.2; extra == "transformer"
Requires-Dist: pyarrow>=15.0.0; extra == "transformer"
Provides-Extra: loader
Requires-Dist: pyarrow>=15.0.0; extra == "loader"
Provides-Extra: test
Requires-Dist: pytest>=9.1.0; extra == "test"
Requires-Dist: pytest-asyncio>=1.4.0; extra == "test"
Requires-Dist: testcontainers[clickhouse]>=4.14.2; extra == "test"
Requires-Dist: clickhouse-connect>=1.3.0; extra == "test"
Requires-Dist: pymongo>=4.9; extra == "test"
Requires-Dist: aioodbc>=0.5.0; extra == "test"
Requires-Dist: asyncpg>=0.31.0; extra == "test"
Requires-Dist: etlpy-tools[all]; extra == "test"
Provides-Extra: all
Requires-Dist: etlpy-tools[extractor,loader,transformer]; extra == "all"
Dynamic: license-file

<p align="center">
  <img src="https://raw.githubusercontent.com/Sque-ak/etlpy/main/resources/logo.png" alt="etlpy" width="360">
</p>

<p align="center">
  <strong>A composable, async toolkit for building ETL pipelines in Python  powered by Polars.</strong>
</p>

<p align="center">
  <img src="https://img.shields.io/badge/python-3.13+-blue.svg" alt="Python 3.13+">
  <img src="https://img.shields.io/badge/engine-Polars-orange.svg" alt="Polars">
  <img src="https://img.shields.io/badge/license-MIT-green.svg" alt="MIT License">
  <img src="https://img.shields.io/badge/status-alpha-yellow.svg" alt="Alpha">
  <img src="https://img.shields.io/badge/coverage-100%25-brightgreen.svg" alt="Coverage 100%">
  <img src="https://img.shields.io/pypi/v/etlpytools.svg" alt="PyPI">
</p>

---

**ETLPY** is a small, unopinionated set of building blocks for extract / transform / load work.
You compose pipelines from tiny, single-purpose **steps** and run them - in a script or inside an Airflow task. It is a _toolkit_, not a framework: take what you need, write your own
steps for the rest.

- **Pipeline** - Just a list of async `Step`s. Add, remove, reorder.
- **Polars-native** - one engine end to end (Arrow under the hood).
- **Async** - steps are `async`, so you can fan out concurrent API/DB calls inside a step.
- **Many databases** - ClickHouse, PostgreSQL, MS SQL Server, MongoDB and SQLite connectors, plus a parquet data lake. Read from one, load into another.
- **Schema-guaranteed** - validate and coerce data against a Pydantic model, so downstream gets exactly the types you declared.
- **Deduplication** - row-hash change detection loads only new or changed rows.
- **Airflow-friendly** - one pipeline = one task. etlpy moves the data; Airflow orchestrates.

## Installation

```bash
pip install etlpy-tools            # core tools kit

# or with the stage extras you need
pip install "etlpy-tools[all]"          # everything
pip install "etlpy-tools[extractor]"    # httpx-based API steps
pip install "etlpy-tools[transformer]"  # Polars transform steps
pip install "etlpy-tools[loader]"       # data lake + DB loaders
```

> Installed as **`etlpy-tools`**, imported as **`etl`** e.g. `from etl.generic import Pipeline`.

Database drivers are **not** bundled install the one you use:
`clickhouse-connect`, `asyncpg`, `aioodbc`, `pymongo`, or `aiosqlite`.

Requires Python 3.13+.

## Quick start

A pipeline threads a Polars `DataFrame` through a list of steps:

```python
import asyncio
from etl.generic import Pipeline, Pipestart
from etl.transformer.steps import ClearText, DropDuplicates, GenerateKey, RowHash
from etl.loader.steps.datalake import Save

@Pipestart
async def clean_customers(raw_df):
    return Pipeline([
        ClearText(),
        DropDuplicates(),
        GenerateKey(columns=["id"], key_name="pk"),
        RowHash(),                             # content fingerprint for change detection
        Save(name="customers", layer="raw"),   # persist to the data lake
    ], dataframe=raw_df)


asyncio.run(clean_customers(raw_df))
```

Load a lake file into ClickHouse, inserting only what actually changed:

```python
from etl.generic import Pipeline
from etl.extractor.steps.datalake import Read
from etl.loader.steps.clickhouse import EnsureTable, Delta, Insert
from etl.extractor.steps.clickhouse import Connect
from etl.loader.steps.datalake import Archive

await Pipeline([
    Connect(host="clickhouse", database="analytics"),
    Read(layer="fact", name="transactions"),
    EnsureTable("fact_transactions",
                engine="ReplacingMergeTree(loaded_at)", order_by=["pk"]),
    Delta("fact_transactions", keys=["pk"]),   # skip rows already loaded unchanged
    Insert("fact_transactions"),
    Archive(layer="fact", name="transactions"),
]).run()
```

Any step can halt the pipeline gracefully by raising `StopPipeline`. It is a
control-flow signal, not an error: `run()` stops, skips the remaining steps and
returns the data produced so far (no traceback).

```python
from etl.generic import Step, StopPipeline

class StopIfEmpty(Step):
    async def apply(self, df, data=None):
        if df is None or df.is_empty():
            raise StopPipeline("no rows")          # stop, return df as-is
        return df
```

Pass `df=` to control what the pipeline returns on stop:

```python
raise StopPipeline("threshold hit", df=partial)
```

## Core concepts

| Piece              | What it is                                                                                                                                                                      |
| ------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **`Pipeline`**     | An ordered list of steps. `run()` threads a DataFrame through them and returns the final one.                                                                                   |
| **`StopPipeline`** | A step may raise StopPipeline to halt the run early                                                                                                                             |
| **`Pipestart`**    | A decorator that runs a pipeline-returning function.                                                                                                                            |
| **`Step`**         | A unit of work: `async def apply(self, df, data) -> df`. Write your own by subclassing.                                                                                         |
| **`Data`**         | A shared context (auth tokens, DB clients, config) passed to every step. Mutated by reference; **DataFrames don't live here** - they flow as the threaded `df` or via the lake. |

The split is deliberate: the **`DataFrame`** is threaded and returned, while the **`Data`** context is
shared state. Heavy tables move through the data lake, not through the context - which keeps each
pipeline a clean fit for a single Airflow task.

## The toolbox

Every step is one small class with an `async def apply(self, df, data) -> df`. Below is the full set.
Missing one? Subclass `Step`, implement `apply`, and drop it into the list.

## Schema guarantee with `ToSchema`

Raw data lies: missing fields, wrong types, extra columns, values that don't parse.
`ToSchema` is the gate that turns messy input into exactly the data you declared. It
validates and coerces **every row against a Pydantic model** and returns a DataFrame whose
schema comes from that model - what is not in the model is dropped, what cannot be coerced
raises. Downstream steps and your warehouse table then get clean, typed data by construction.

Declare the canon as a Pydantic model:

```python
from datetime import date
from pydantic import BaseModel

class Transaction(BaseModel):
    id: int
    amount: float
    currency: str
    txn_date: date
    note: str | None = None          # optional -> nullable column
```

Drop `ToSchema` into a pipeline:

```python
from etl.transformer.steps import ToSchema, GenerateKey, RowHash

Pipeline([
    # ... read raw data ...
    ToSchema(Transaction, mapping={"trans_id": "id", "sum": "amount"}),  # rename + validate + type
    GenerateKey(columns=["id"], key_name="pk"),
    RowHash(),
    # ... load ...
])
```

What the step guarantees:

- **Exact schema & types** - columns and dtypes come from the model, not from guessing on the
  first N rows: `int` -> `Int64`, `date` -> `Date`, `str | None` -> nullable `String`, and so on.
- **Extra columns dropped** - anything outside the model is discarded.
- **Fail fast** - a value that can't be coerced (or a missing required field) raises, so bad
  upstream data never silently reaches the warehouse.
- **`mapping`** - rename raw source fields to the canonical names (1:1) before validation.

`ToSchema` never imports pydantic itself - it only uses the model you pass, so **pydantic
stays an optional, user-side dependency**. It also works with
[Patito](https://github.com/JakobGM/patito) models (Pydantic on Polars). Validation runs per
row - thorough rather than fast; correctness is the point, speed is recovered downstream.

### Extract

- **`OAuthenticate(url, credentials, fields=OAuthFields(...), send="json", method="POST", auth_header=None, headers=None, store="auth", timeout=60.0)`** - non-interactive OAuth 2.0 token flow (client-credentials). Sends `credentials` (JSON or form via `send`), pulls the token out of the response by dotted paths (`fields`), applies `auth_header` (e.g. `{"Authorization": "Bearer {token}"}`, `{token}` is filled in) to the shared httpx client, and stores the parsed auth in `data["auth"]`. Request-level headers (gateway API keys, `Content-Type`) go in `headers`.
- **`AuthenticateBasic(user, password, headers=None, timeout=60.0)`** _(HTTP Basic, RFC 7617)_ - sets `httpx.BasicAuth(user, password)` on the shared client so every downstream request carries `Authorization: Basic ...`. No token exchange - the header is static.
- **`Read(layer, name)`** _(data lake)_ - read `{name}.parquet` from a lake layer into the pipeline df.
- **`Read(query)`** _(ClickHouse)_ - run a SQL query (client from `data["ch"]`) and return the result as a Polars frame.
- **`Reads(layer, name, mode=STATIC, date_from=None, date_to=None, pattern="*.parquet", missing_ok=False)`** _(data lake)_ - read and concat every parquet in a lake folder, with an optional inclusive date range on the file name.
- **`Connect(host, port=8123, database, username, password, secure=False)`** _(ClickHouse)_ - open a clickhouse-connect client (extra kwargs are forwarded) and store it in `data["ch"]`.

> **`OAuthFields`** maps where each token-response field lives, e.g.
> `OAuthFields(access_token="data.access_token", expires_in="data.expires_in")`.
> Set a field to `None` if the API does not return it.

### Transform

- **`AddColumn(column_name, value)`** - add a constant-value column.
- **`CastTypes({column: dtype})`** - cast columns to Polars dtypes (pin types before loading).
- **`RenameColumns({old: new})`** - rename columns.
- **`DropColumns(columns, exclude=False)`** - drop the listed columns; `exclude=True` keeps only them.
- **`DropDuplicates(subset=None)`** - drop duplicate rows (by `subset`, or all columns).
- **`DropNulls(subset=None)`** - drop rows with a null in any of `subset`.
- **`FillNulls(value, columns=None)`** - fill nulls; `value` is a scalar or a `{column: value}` dict.
- **`FilterRows(expr)`** - keep rows matching a Polars boolean expression, e.g. `pl.col("amount") > 0`.
- **`TrimString(columns=None)`** - strip whitespace from string columns (`None` / `"*"` = all string columns).
- **`ClearText(columns="*")`** - clean text: collapse newlines to spaces, drop quotes/backslashes, trim.
- **`NormalizeNumeric(columns, method="minmax")`** - scale numeric columns (`"minmax"` or `"zscore"`).
- **`GenerateKey(columns, key_name="pk", mode="hash")`** - build a key column. `mode`: `"hash"` (SHA-256 hex), `"hash_int"` (stable 64-bit int), `"sequential"` (1, 2, 3, ...).
- **`RowHash(exclude=None, separator="||")`** - SHA-256 fingerprint of every row value (minus `row_hash`, `loaded_at`, and `exclude`) written to `row_hash`, for change detection.
- **`Aggregate(group_by, aggregations)`** - group + aggregate, e.g. `{"amount": ["sum", "mean"]}`; output columns are `{column}_{func}`.
- **`Join(other, on, how="inner", select=None, prefix=None)`** - join with another frame **or a sub-`Pipeline`** (run to produce the right side); `select` / `prefix` shape the right columns.
- **`ExtractEntities(sources, defaults=None)`** - stack several column groups into one long table (e.g. sender/receiver columns into a single `party` table).
- **`ToSchema(model, mapping=None)`** - validate and coerce the df against a Pydantic model; output has the model's exact schema (see [Schema guarantee](#schema-guarantee-with-toschema)).
- **`Union(other, how="vertical")`** - stack another DataFrame **or a sub-`Pipeline`** onto the df (`vertical` / `diagonal` / `horizontal`).
- **`SQL(query, view_name="source")`** - run a Polars-SQL query over the df (registered as `view_name`); `SQL.from_file(path)` loads the query from a `.sql` file.
- **`Lambda(func)`** - apply an arbitrary `df -> df` callable (escape hatch for one-off logic).

### Load

- **`EnsureTable(table, order_by, engine="MergeTree", partition_by=None, if_exists="append")`** - create `table` from the df schema if it does not exist. `order_by` is required (it is the primary / dedup key); `if_exists="error"` raises when the table already exists.
- **`Delta(table, keys)`** - keep only new or changed rows by comparing `row_hash` against the table (needs a `ReplacingMergeTree`-family engine). The core of idempotent loads.
- **`Insert(table)`** - insert the df into a ClickHouse table via Arrow (an empty / `None` frame is skipped).
- **`Save(name, layer="raw")`** _(data lake)_ - write the df to a lake layer as `{name}.parquet`, overwriting in place.
- **`Optimize(table)`** _(ClickHouse)_ - `OPTIMIZE TABLE ... FINAL` to collapse ReplacingMergeTree duplicates (expensive; run after a load, not per row).
- **`Archive(layer, name)`** _(data lake)_ - move a lake file into the `archive` layer (e.g. after a successful load).

## Connectors

etlpy ships steps for several databases. **Drivers are not bundled** - there are many
databases, so you install only the one you need. Each `Connect` opens a client and stores it
in `data[...]`; the read/load steps pick it up from there.

| Database          | Steps                                                           | `data` key       | Driver (install yourself) |
| ----------------- | --------------------------------------------------------------- | ---------------- | ------------------------- |
| **ClickHouse**    | `Connect`, `Read`, `EnsureTable`, `Delta`, `Insert`, `Optimize` | `data["ch"]`     | `clickhouse-connect`      |
| **PostgreSQL**    | `Connect`, `Read`, `EnsureTable`, `Insert`                      | `data["pg"]`     | `asyncpg`                 |
| **MS SQL Server** | `Connect`, `Read`, `EnsureTable`, `Insert`                      | `data["mssql"]`  | `aioodbc` (+ ODBC driver) |
| **MongoDB**       | `Connect`, `Read`, `EnsureTable`, `Delta`, `Upsert`             | `data["mongo"]`  | `pymongo` (async)         |
| **SQLite**        | `Connect`, `Read`, `EnsureTable`, `Insert`                      | `data["sqlite"]` | `aiosqlite`               |

Import `Connect`/`Read` from `etl.extractor.steps.<db>` and the write steps from
`etl.loader.steps.<db>`. SQL reads are parameterized (injection-safe): ClickHouse
`{name:Type}`, Postgres `$1`, MS SQL / SQLite `?`; MongoDB uses a query dict.

> **SQLite** is meant for tests and small local datasets (great as an in-memory `:memory:`
> backend). It is a single-file, single-writer engine and is **not suitable for storing large
> data** - use ClickHouse or Postgres for that.

## Testing

Every step has its own unit tests, and the whole extract / transform / load chain is exercised end to
end against a real ClickHouse (spun up in Docker via `testcontainers`). A complete pipeline example
lives in [`generic/tests/test_pipeline.py`](src/etl/generic/tests/test_pipeline.py).

```bash
pip install -e ".[all,test]"
pytest -q
```

<p align="center"> <img src="https://raw.githubusercontent.com/Sque-ak/etlpy/main/resources/100cov.png" alt="100% coverage" width="640"> </p>

## Status

Alpha, under active development. A PyPI release is planned.

## License

Released under the [MIT License](LICENSE).
