Metadata-Version: 2.4
Name: pypelite
Version: 0.1.8
Summary: Opinionated pipeline run archives for Python scripts.
Author: pypelite contributors
License-Expression: MIT
Project-URL: Documentation, https://edgy-raven.github.io/pypelite/
Project-URL: Source, https://github.com/edgy-raven/pypelite
Keywords: pipeline,cache,workflow,data
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: docs
Requires-Dist: sphinx; extra == "docs"
Dynamic: license-file

# pypelite

Pypelite is an opinionated pipeline layer in the spirit of
[joblib](https://joblib.readthedocs.io/).

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.

```python
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

```sh
pip install pypelite
```

## Documentation

[API Reference](https://edgy-raven.github.io/pypelite/)

## Run Controls

```python
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:

```python
@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:

```python
@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:

```python
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.
