pypelite 0.1.0

Tiny pipeline library for ordinary Python scripts.

pypelite

Pypelite is a tiny pipeline library for ordinary Python scripts.

import pypelite

@pypelite.stage("load", batch="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)

Pipelines are resumable: completed stages load from disk, and selected stages can be refreshed when they need to run again. No DSL, no DAG boilerplate, no Airflow deployment. The Python code is the pipeline.

Installation

pip install pypelite

Refresh, Skip, Clean

Control a run from the pipeline context.

with pypelite.pipeline(
    "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.

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

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

Fanout

Use workers when fanout should run in parallel.

@pypelite.stage("features", key="symbol", batch="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 batch when work should run in chunks.

@pypelite.stage("predict", key="symbol", batch="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 assemble one final artifact; keyed batches save one result 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.