pypelite 0.1.9

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.

Cache Identity

Stages keep one artifact by default. Set key=True when all call arguments should identify separate artifacts:

@pypelite.stage("features", key=True)
def build_features(symbol, window):
    return feature_builder.for_symbol(symbol, window)

String, tuple, and callable keys select an explicit identity. Add source=True when changing the function source should create a new artifact:

@pypelite.stage("features", key="symbol", source=True)
def build_features(symbol, window):
    return feature_builder.for_symbol(symbol, window)

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)

Set reject_none=True when a missing item output should fail the run instead of being cached as None.

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": pypelite.Archive(
            "archive/data", formatters=[pypelite.ArchiveFormat()]
        ),
    },
):
    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.