Metadata-Version: 2.4
Name: paralify
Version: 0.1.0
Summary: Parallelize a function over pandas DataFrames, lists, tuples, sets and dicts — with retries, periodic checkpoints and crash-safe resume.
Author: hrosmendez
License-Expression: MIT
Project-URL: Homepage, https://pypi.org/project/paralify/
Keywords: parallel,threads,checkpoint,resume,dataframe,pandas
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: tqdm
Provides-Extra: pandas
Requires-Dist: pandas; extra == "pandas"
Dynamic: license-file

# paralify

**Parallelize any Python function over a pandas DataFrame, list, tuple, set or dict — with automatic retries, periodic checkpoints and crash-safe resume.**

`paralify` is built for long-running batch jobs (API calls, LLM requests, scraping, heavy computations) where a crash halfway through should **never** mean starting over. You can `run()` a job from scratch, and if anything goes wrong — a network outage, a killed process, a laptop reboot — just call `resume()` and it picks up exactly where it left off, reprocessing only the items that are still pending or that exhausted their retries.

## Features

- ⚡ **Parallel execution** with a configurable thread pool.
- 🔁 **Automatic retries** per item, with a configurable wait between attempts.
- 💾 **Periodic checkpoints** every `save_every` processed items (set it to `None` to disable checkpoints entirely).
- 🛟 **Crash-safe `resume()`**: recovers progress from the latest checkpoint (or the final output file) and only processes what is missing.
- 📦 **Works on DataFrames, lists, tuples, sets and dicts** with a single, unified API.
- 📊 Progress bar via `tqdm`.
- Results always keep the **same order** as the input (and the same keys for dicts).

## Installation

```bash
pip install paralify
```

If you want to process pandas DataFrames:

```bash
pip install paralify[pandas]
```

## Quick start: `run()`

```python
from paralify import ParallelProcessor

def square(x):
    return x * x

processor = ParallelProcessor(
    data=[1, 2, 3, 4, 5],
    process_fn=square,
    output_file="results.pkl",
    n_workers=4,
)

results = processor.run()
# [1, 4, 9, 16, 25]  — same order as the input, also saved to results.pkl
```

## Crash-safe `resume()`

This is the core feature of `paralify`. Imagine a long job calling an API for 100,000 rows:

```python
import pandas as pd
from paralify import ParallelProcessor

def call_api(row, model="gpt"):
    # Any function that may fail intermittently
    return expensive_api_call(row["prompt"], model=model)

df = pd.DataFrame({"prompt": [...]})  # 100,000 rows

processor = ParallelProcessor(
    data=df,                      # each row is passed to call_api as a dict
    process_fn=call_api,
    output_file="api_results.pkl",
    n_workers=8,
    max_retries=5,
    retry_wait_seconds=5,
    save_every=1000,              # checkpoint every 1000 finished items
    process_kwargs={"model": "gpt"},
)

results = processor.run()
```

If the process dies at item 73,421 — power cut, OOM, Ctrl+C, whatever — you do **not** start over. In a fresh Python session:

```python
processor = ParallelProcessor(
    data=df,                      # same input as before
    process_fn=call_api,
    output_file="api_results.pkl",  # same output file
    save_every=1000,
)

results = processor.resume()
# Loading progress from: api_results_checkpoint_73000.pkl
# 73000/100000 items already done
# 27000 items left to (re)process
```

`resume()` also retries items that failed all their attempts during a previous run, so you can call it as many times as needed until everything succeeds. Calling it when everything is already done is a no-op that simply returns the results.

## Works with lists, tuples, sets and dicts too

The same API works on any supported collection — `process_fn` receives one element at a time:

```python
# dict: the function receives each VALUE, results keep the original KEYS
urls = {"home": "https://a.com", "blog": "https://b.com"}
processor = ParallelProcessor(urls, fetch, output_file="pages.pkl")
pages = processor.run()
# {"home": <result>, "blog": <result>}

# tuple / set
processor = ParallelProcessor((10, 20, 30), square, output_file="out.pkl")
processor = ParallelProcessor({"a", "b", "c"}, process, output_file="out.pkl")
```

> **Note on sets:** sets are unordered, so `paralify` sorts their elements when
> possible to get a deterministic order — this is what makes `resume()` safe
> across different Python sessions.

## Disabling checkpoints

If your job is short or you don't need crash recovery, set `save_every=None` (or `0`) and no checkpoint files will be created — only the final output file:

```python
processor = ParallelProcessor(
    data=my_list,
    process_fn=my_fn,
    output_file="out.pkl",
    save_every=None,   # no checkpoints
)
```

## Error handling

Each item is retried up to `max_retries` times. If every attempt fails, the result for that item is a dict of the form `{"error": "<message>", "item": <original item>}` instead of crashing the whole job. Those items are considered *pending*, so a later `resume()` will try them again.

## API reference

```python
ParallelProcessor(
    data,                      # DataFrame | list | tuple | set | dict
    process_fn,                # called as process_fn(item, **process_kwargs)
    output_file,               # final results pickle path
    n_workers=6,               # thread pool size
    max_retries=5,             # attempts per item
    retry_wait_seconds=5,      # wait between attempts
    save_every=1000,           # checkpoint frequency; None/0 disables
    desc="Processing",         # tqdm progress bar label
    process_kwargs=None,       # extra kwargs forwarded to process_fn
)

processor.run()     # process everything from scratch
processor.resume()  # continue from the latest checkpoint / output file
```

## License

MIT
