Documentation

Install the library and add two lines to your pipeline: ```bash pip install "pipelinehub>=0.1.11" ``` ```python import pandas as pd from pipelinehub import DataPipeline pipeline = DataPipeline(name="my-pipeline") @pipeline.add_step def extract(df): return pd.read_csv("data.csv") @pipeline.add_step def clean(df): return df.dropna(subset=["id"]) pipeline.execute(pd.DataFrame()) ``` Set your API key to sync runs to the cloud dashboard: ```bash export PIPELINEHUB_API_KEY=ph_live_xxx ```

`@pipeline.add_step` works in three forms: ```python # 1 — bare decorator @pipeline.add_step def extract(df): ... # 2 — named step (used for SLA tracking continuity) pipeline.add_step(my_fn, name="reconcile") # 3 — explicit call (reuse across multiple pipelines) pipeline2.add_step(extract) ``` Because `add_step` returns the original function, decorated steps stay reusable: ```python pipeline1 = DataPipeline(name="etl") pipeline2 = DataPipeline(name="etl") @pipeline1.add_step def validate(df): return df.dropna(subset=["id"]) pipeline2.add_step(validate) # same function, second pipeline ```

pipelinehub works with polars DataFrames and LazyFrames natively (>= 0.1.11). LazyFrames are collected automatically before profiling and before passing to the next step — no `.collect()` needed in your code: ```python import polars as pl from pipelinehub import DataPipeline pipeline = DataPipeline(name="finance-etl") @pipeline.add_step def ingest(df: pl.DataFrame) -> pl.LazyFrame: return pl.scan_csv("transactions.csv") # LazyFrame — auto-collected @pipeline.add_step def clean(df: pl.DataFrame) -> pl.DataFrame: # receives collected DataFrame return df.drop_nulls(subset=["account_id"]) ```

pipelinehub automatically compares each run against the previous successful run and flags: | Anomaly | Trigger | |---|---| | `row_count_change` | Row count changes >50% | | `null_increase` | Null % increases >20pp | | `schema_change` | Columns added, removed, or retyped | | `slow_step` | Step takes >3× its rolling average duration | No configuration required. All anomalies appear in your dashboard.

pipelinehub tracks rolling average duration per step name. A `slow_step` anomaly fires when a step takes more than 3× its historical average. Use `name=` to maintain SLA continuity when a function is renamed: ```python def reconcile_v2(df): # new implementation return df.group_by("account_id").agg(...) # name="reconcile" → compared against prior "reconcile" baseline pipeline.add_step(reconcile_v2, name="reconcile") ```

Set `PIPELINEHUB_API_KEY` in your environment — every run syncs to the dashboard automatically: ```bash export PIPELINEHUB_API_KEY=ph_live_xxx python3 pipeline.py ``` Or pass it to the constructor: ```python pipeline = DataPipeline( name="etl", api_key="ph_live_xxx", api_url="https://api.pipelinehub.cloud", # default ) ``` **If no key is set**, pipelinehub runs entirely locally using SQLite — no errors, no cloud calls. The OSS library is always free.