pypelite 0.1.11

Opinionated pipeline run archives for Python scripts.

pypelite

Pypelite builds expressive pipelines from ordinary Python. It provides more structure than joblib without requiring an Airflow-style DAG or orchestration platform.

See the API reference for exact options. Pypelite is designed with AI-assisted use in mind; see AGENTS.md for generation guidance.

Install

pip install pypelite

Pipeline

Decorate pipeline steps, then call them inside pypelite.pipeline(...):

import pypelite

@pypelite.checkpoint()
def load_records(path):
    return read_records(path)

@pypelite.checkpoint()
def build_features(records_df):
    return make_model_features(records_df)

@pypelite.checkpoint()
def train_model(features_df):
    return fit_price_model(features_df)

with pypelite.pipeline("runs/price-model"):
    records_df = load_records("records.parquet")
    features_df = build_features(records_df)
    model = train_model(features_df)

A failed run can resume from completed steps because outputs are stored in the pipeline archive.

Pipeline controls are ordinary context options:

with pypelite.pipeline(
    "runs/experiment",
    refresh=["build_features"],
    skip=["train_model"],
    clean=["predict"],
    until="build_features",
):
    run_price_model()

Archive Management

Each cached function owns one directory under its archive:

archive/
├── load_records/
│   └── artifact.pkl
└── load_price/
    └── AAPL~7d3a4c1f2b80.pkl

Checkpoints store one result, while stages store keyed results. Vectorized and batched stages use the same per-item layout; see Vectorization and Batching.

Named archives let pipelines share durable data:

market = pypelite.Archive("archives/market")

@pypelite.stage(archive="market")
def load_price(symbol):
    return market_api.price(symbol)

with pypelite.pipeline("runs/model-a", archives={"market": market}):
    aapl = load_price("AAPL")

with pypelite.pipeline("runs/model-b", archives={"market": market}):
    aapl = load_price("AAPL")  # reused from the shared archive

Optional archive configurations support native formats for common tables, arrays, and models. Additional archives supplied as plain paths receive the pickle fallback.

Vectorization and Batching

Both stage collection modes cache one result per item. vectorize adapts a function that accepts one item so callers can pass a collection:

@pypelite.stage(vectorize="symbol", workers=4)
def load_price(symbol):
    return market_api.price(symbol)

batch is for a function that already accepts a collection. Missing items are grouped into worker calls:

@pypelite.stage(key="symbol", batch="symbols", workers=4)
def load_prices(symbols):
    return [market_api.price(symbol) for symbol in symbols]

A checkpoint can also run in batches when the combined output should remain one result:

@pypelite.checkpoint(batch="records", batch_size=50, workers=4)
def score_dataset(records):
    return model.score(records)