pypelite 0.1.8

Opinionated pipeline run archives for Python scripts.

pypelite

Pypelite is an opinionated pipeline layer in the spirit of joblib.

Wrap expensive functions with @pypelite.stage(...) and run your script inside pypelite.pipeline(...). Stage outputs are saved to disk, so a failed run can resume from completed steps, and a later run can refresh only the stages you want to rerun.

import pypelite

@pypelite.stage("prices")
def load_prices(symbols):
    return market_api.fetch_prices(symbols)

@pypelite.stage("features")
def build_features(prices_df):
    return make_model_features(prices_df)

@pypelite.stage("train")
def train_model(features_df):
    return fit_price_model(features_df)

with pypelite.pipeline("runs/price-model"):
    prices_df = load_prices(["AAPL", "MSFT", "NVDA"])
    features_df = build_features(prices_df)
    model = train_model(features_df)

No DSL, no DAG boilerplate, no scheduler deployment. The Python script is the pipeline.

Installation

pip install pypelite

Documentation

API Reference

Run Controls

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

• refresh recomputes named stages touched by the run.
• skip returns a stage's skip_value.
• clean removes cached files not touched by a successful run.
• until stops the pipeline after the named stage completes.

Parallel Runs

Use vectorize with keyed stages when workers should compute missing items in batches while still storing one artifact per key. A common case is filling missing dates:

@pypelite.stage("daily_features", key="date", vectorize="dates", workers=4)
def build_daily_features(dates):
    return feature_builder.for_dates(dates)

Use parallel_batch when the whole stage should run in worker-backed batches and produce one final cached artifact. A common case is gathering many stocks:

@pypelite.stage(
    "prices",
    parallel_batch="symbols",
    batch_size=50,
    workers=4,
)
def load_prices(symbols):
    return market_api.fetch_prices(symbols)

Archive Management

Stages write to the default archive unless they name another one. Keep shared data outside the training job archive:

import pypelite.configs

@pypelite.stage("prices", archive="data", key=("date", "symbol"))
def load_price(date, symbol):
    return market_api.price(date, symbol)

@pypelite.stage("model")
def train_model(features_df):
    return xgboost.train(params, features_df)

with pypelite.pipeline(
    archives={
        "default": pypelite.configs.archive("runs/model"),
        "data": "archive/data",
    },
):
    run_price_model()

Use custom formatters when a stage should write a native artifact format. The model stage above returns an XGBoost booster, so pypelite.configs.archive stores it as UBJ.