Metadata-Version: 2.4
Name: pypelite
Version: 0.1.0
Summary: Tiny pipeline library for ordinary Python scripts.
Author: pypelite contributors
License-Expression: MIT
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
Dynamic: license-file

# pypelite

_Pypelite is a tiny pipeline library for ordinary Python scripts._

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

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

```sh
pip install pypelite
```

## Refresh, Skip, Clean

Control a run from the pipeline context.

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

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

```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
assemble one final artifact; keyed batches save one result 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.
