Metadata-Version: 2.4
Name: pypelite
Version: 0.1.4
Summary: Tiny pipeline library for ordinary 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>=8; extra == "docs"
Dynamic: license-file

# 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.

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

No DSL, no DAG boilerplate, no Airflow deployment. The Python code is the
pipeline.

## Installation

```sh
pip install pypelite
```

## Documentation

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

## Refresh, Skip, Clean

Control a run from the pipeline context.

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

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

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

```python
@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 use
temporary chunk files while building one final artifact; keyed batches save one
result file per key.

### Shared Archives

Stages can choose named archives.

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