Metadata-Version: 2.4
Name: pypelite
Version: 0.1.10
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 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](https://github.com/edgy-raven/pypelite/blob/main/AGENTS.md) for
canonical generation guidance.

```sh
pip install pypelite
```

## Pipeline

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

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

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

```text
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](#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](#vectorization-and-batching).

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

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

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

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

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

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

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

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