pypelite 0.1.10

Opinionated pipeline run archives for Python scripts.

pypelite

Pypelite adds persistent checkpoints and keyed stages to ordinary Python scripts. Python remains the pipeline definition; pypelite supplies archives, parallel collection handling, and run controls.

The compact API and auditable archive layout are designed with AI-assisted coding in mind. See AGENTS.md for canonical generation guidance.

pip install pypelite

Pipeline

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

import pypelite

@pypelite.checkpoint()
def load_prices(symbols):
    return market_api.fetch_prices(symbols)

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

@pypelite.checkpoint()
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 DAG or scheduler is constructed. 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()

refresh, skip, and clean take lists or tuples of cached-function names. until names the final step to run.

Archive Management

Each cached function owns one directory under its archive:

archive/
├── .pypelite.lock
├── load_prices/
│   ├── artifact.pkl
│   └── meta.json
└── features/
    ├── AAPL~7d3a4c1f2b80.pkl
    └── meta.json

The checkpoint and stage filename patterns are explained in Checkpoints and Stages. meta.json maps full cache identities to filenames and records producing arguments. Vectorized and batched stages use this same per-item layout; see Vectorization and Batching.

Pass the default archive path positionally. Configured archives use archive=, while archives= contains only additional named archives:

import pypelite.configs

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

@pypelite.checkpoint()
def train_model(features_df):
    return xgboost.train(params, features_df)

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

Writable pipelines hold exclusive archive locks. read_only=True uses shared locks so multiple readers can run together; cache misses and other writes then raise. Pipelines using unrelated archives can run concurrently.

Checkpoints and Stages

A checkpoint owns one artifact. Its call arguments are recorded but do not select separate results:

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

A stage owns one artifact per call identity. All arguments form the key by default; a string, tuple, or callable selects a narrower key:

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

Stage filenames contain up to 20 characters of formatted key values followed by a hash. Complete values and field names remain in meta.json. Arguments do not need to be JSON-serializable: pypelite hashes each value with its archive formatter, falling back to pickle.

Set reject_changed=True when a hit should fail if non-key arguments differ. Set source=True when source changes should intentionally create a new artifact. Use pypelite.configs.archive(...) when values should use native formats such as CSV, NumPy NPZ, XGBoost UBJ, or Keras weights.

Vectorization and Batching

Both stage collection modes retain one cached artifact per item. vectorize adapts a function that accepts one item so callers can pass a collection:

@pypelite.stage(vectorize="date", workers=4)
def build_daily_feature(date):
    return feature_builder.for_date(date)

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

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

With vectorize, key selects from scalar call arguments. With batch, it selects from each item in the collection.

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

@pypelite.checkpoint(batch="symbols", batch_size=50, workers=4)
def load_prices(symbols):
    return market_api.fetch_prices(symbols)

API Reference