Metadata-Version: 2.4
Name: pypelite
Version: 0.1.11
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: Operating System :: POSIX
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 builds expressive pipelines from ordinary Python. It provides more
structure than [joblib](https://joblib.readthedocs.io/) without requiring an
[Airflow](https://airflow.apache.org/)-style DAG or orchestration platform.

See the [API reference](https://edgy-raven.github.io/pypelite/) for exact
options. Pypelite is designed with AI-assisted use in mind; see
[AGENTS.md](https://github.com/edgy-raven/pypelite/blob/main/AGENTS.md) for
generation guidance.

## Install

```sh
pip install pypelite
```

## Pipeline

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

```python
import pypelite

@pypelite.checkpoint()
def load_records(path):
    return read_records(path)

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

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

with pypelite.pipeline("runs/price-model"):
    records_df = load_records("records.parquet")
    features_df = build_features(records_df)
    model = train_model(features_df)
```

A failed run can resume from completed steps because outputs are stored in the
pipeline archive.

Pipeline controls are ordinary context options:

```python
with pypelite.pipeline(
    "runs/experiment",
    refresh=["build_features"],
    skip=["train_model"],
    clean=["predict"],
    until="build_features",
):
    run_price_model()
```

## Archive Management

Each cached function owns one directory under its archive:

```text
archive/
├── load_records/
│   └── artifact.pkl
└── load_price/
    └── AAPL~7d3a4c1f2b80.pkl
```

Checkpoints store one result, while stages store keyed results. Vectorized and
batched stages use the same per-item layout; see
[Vectorization and Batching](#vectorization-and-batching).

Named archives let pipelines share durable data:

```python
market = pypelite.Archive("archives/market")

@pypelite.stage(archive="market")
def load_price(symbol):
    return market_api.price(symbol)

with pypelite.pipeline("runs/model-a", archives={"market": market}):
    aapl = load_price("AAPL")

with pypelite.pipeline("runs/model-b", archives={"market": market}):
    aapl = load_price("AAPL")  # reused from the shared archive
```

Optional archive configurations support native formats for common tables,
arrays, and models. Additional archives supplied as plain paths receive the
pickle fallback.

## Vectorization and Batching

Both stage collection modes cache one result per item. `vectorize`
adapts a function that accepts one item so callers can pass a collection:

```python
@pypelite.stage(vectorize="symbol", workers=4)
def load_price(symbol):
    return market_api.price(symbol)
```

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

```python
@pypelite.stage(key="symbol", batch="symbols", workers=4)
def load_prices(symbols):
    return [market_api.price(symbol) for symbol in symbols]
```

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

```python
@pypelite.checkpoint(batch="records", batch_size=50, workers=4)
def score_dataset(records):
    return model.score(records)
```
