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

pip install paralify

If you want to process pandas DataFrames:

pip install paralify[pandas]

Quick start: run()

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:

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:

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:

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

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

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