pypelite 0.1.5

Tiny pipeline library for ordinary Python scripts.

pypelite

Pypelite is a tiny pipeline library for ordinary Python scripts.

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

import pypelite

@pypelite.stage("load", fanout_arg="symbols", batch_size=50, workers=4)
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 Airflow deployment. The Python code is the pipeline.

Installation

pip install pypelite

Documentation

• Documentation

Refresh, Skip, Clean

Control a run from the pipeline context.

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 without touching its cache.
• clean removes old keyed artifacts not touched by a successful run.
• until stops after the named stage completes.

Advanced Features

Item Caches

Use key when each item should have its own saved result. Use ignore to leave fields out of a composite key while still passing them to the function.

@pypelite.stage("predict", key=("symbol", "run_id"), ignore="run_id")
def predict_price(symbol, run_id, feature_row):
    return model.predict(feature_row)

with pypelite.pipeline("runs/predictions"):
    for symbol, feature_row in features_df.iterrows():
        predict_price(symbol, run_id="daily", feature_row=feature_row)

Fanout

Use workers when fanout should run in parallel.

@pypelite.stage("features", key="symbol", fanout_arg="symbols", workers=4)
def build_symbol_features(symbols):
    prices_df = market_api.fetch_prices(symbols)
    return make_model_features_by_symbol(prices_df)

with pypelite.pipeline("runs/features"):
    feature_rows = build_symbol_features(["AAPL", "MSFT", "NVDA"])

Batches

Use fanout_arg when work should run in chunks.

@pypelite.stage("predict", key="symbol", fanout_arg="rows", batch_size=200)
def predict_prices(rows):
    return model_api.batch_predict(rows)

with pypelite.pipeline("runs/predictions"):
    predictions = predict_prices(feature_rows)

Batch stages can run chunks in parallel with workers. Unkeyed batches use temporary chunk files while building one final artifact; keyed batches save one result file per key.

Shared Archives

Stages can choose named archives.

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

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

Shared archives make it easy for many experiments to reuse the same market data while keeping model outputs in their own run directories.