Metadata-Version: 2.4
Name: pyarallel
Version: 0.10.0
Summary: The fan-out layer for rate-limited APIs — parallel map with rate limiting, retry, and structured errors. Sync and async, zero dependencies.
Project-URL: Homepage, https://github.com/oneryalcin/pyarallel
Project-URL: Repository, https://github.com/oneryalcin/pyarallel.git
Project-URL: Documentation, https://oneryalcin.github.io/pyarallel
Author-email: Mehmet Oner Yalcin <oneryalcin@gmail.com>
License: MIT License
        
        Copyright (c) 2025 Pyarallel Contributors
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE.md
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown

# Pyarallel

[![PyPI version](https://img.shields.io/pypi/v/pyarallel)](https://pypi.org/project/pyarallel/) [![PyPI Downloads](https://static.pepy.tech/badge/pyarallel/month)](https://pepy.tech/project/pyarallel)

The fan-out layer for rate-limited APIs — one function over many inputs, with rate limiting, retry, resume, and structured errors. Sync and async.

Fanning out over a service that throttles you — LLM calls, embeddings, scraping, any SaaS API — means the same hand-rolled stack every time: a semaphore, tenacity, a token bucket, ad-hoc 429 handling, and a "TODO: resume" you never get to. Pyarallel is that stack, already built and tested. Not DAGs, not queues, not a distributed system — just `concurrent.futures` and `asyncio` with the policies and result handling already built in.

**Zero dependencies. Python 3.12+** — free-threaded 3.13t/3.14t tested, sub-interpreter executor on 3.14.

## Before / After

Fetch 10,000 URLs with rate limiting and error handling.

**concurrent.futures:**

```python
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url):
    return requests.get(url, timeout=10).json()

urls = ["https://api.example.com/users/1", "https://api.example.com/users/2", ...]

results = [None] * len(urls)
errors = []

with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(fetch, url): i for i, url in enumerate(urls)}
    for f in as_completed(futures):
        i = futures[f]
        try:
            results[i] = f.result()
        except Exception as e:
            errors.append((i, e))

# No rate limiting. No retry. No resume. And you still
# need to wire those yourself every time.
```

**pyarallel:**

```python
from pyarallel import parallel_map, RateLimit, Retry

result = parallel_map(
    fetch, urls,
    workers=10,
    rate_limit=RateLimit(100, "minute"),
    retry=Retry(attempts=3, on=(ConnectionError, TimeoutError)),
)

for idx, val in result.successes():
    save(val)
for idx, exc in result.failures():
    log_error(idx, exc)
```

Same thing, async:

```python
import httpx
from pyarallel import async_parallel_map, RateLimit, Retry

client = httpx.AsyncClient()  # ONE client: pooled connections across all calls

async def fetch_async(url):
    return (await client.get(url, timeout=10)).json()

result = await async_parallel_map(
    fetch_async, urls,
    concurrency=10,
    rate_limit=RateLimit(100, "minute"),
    retry=Retry(attempts=3, on=(ConnectionError, TimeoutError)),
)
await client.aclose()
# Same result model — result.ok, result.successes(), result.failures()
```

## Install

```bash
pip install pyarallel
```

## What You Get

- **Rate limiting** — token bucket with burst, per-second/minute/hour: `rate_limit=RateLimit(100, "minute", burst=20)`
- **Shared quota** — one `Limiter` instance across calls and functions when the budget belongs to an API key: `rate_limit=Limiter(RateLimit(100, "minute"))`
- **Retry with backoff** — per-item, exponential, jitter, exception filtering: `retry=Retry(attempts=3, on=(ConnectionError,))`
- **Server-driven backoff** — `retry=Retry.for_http(on=(httpx.HTTPStatusError,))`: 429/503 + `Retry-After` (numeric *and* HTTP-date form) prewired, no client import; the wait also pauses the shared limiter so one throttled task slows the whole pool. Custom policies via `retry_if=`/`wait_from=`
- **Checkpoint/resume** — `checkpoint="run.ckpt"`: a crash at item 40,000 resumes instead of restarting from zero; `checkpoint_key=lambda u: u.id` keys rows by identity so evolving inputs keep completed work
- **Early abort** — `max_errors=10`: a dead API costs tens of calls, not thousands; unrun items are marked `Aborted`, partial results returned
- **Cooperative stop** — `stop=StopToken()`: SIGTERM/notebook-stop lands the plane — admission ceases, checkpoint rows kept, `RunStatus.CANCELLED` reported
- **One windowed engine** — every API (collected and streaming, sync and async) admits work through a bounded in-flight window: lazy input, generators never materialized, no batch barriers, a straggler never stalls the items behind it
- **Streaming** — `parallel_iter` / `async_parallel_iter`: `ordered=True` for input-order yields, per-item `attempts`/`duration`
- **Async sources** — async cursors and paginated generators feed `async_parallel_*` directly, with backpressure to the producer — no draining into a list first
- **Structured errors** — `ParallelResult` with `.ok`, `.ok_values()`, `.successes()`, `.failures()`, `.raise_on_failure()`, plus `.status` (`RunStatus`) for how the run ended; a truncated run is never `.ok` and won't hand out a partial list as if it were complete
- **Timeouts** — total wall-clock on sync *and* async (`timeout=30.0`), per-task in async (`task_timeout=5.0`)
- **Debug mode** — `sequential=True` runs inline: no pool, real stack traces, working breakpoints
- **Progress callbacks** — `on_progress=lambda done, total: print(f"{done}/{total}")` on collected and streaming APIs
- **Process executor** — CPU-bound work: `executor="process"`, with `worker_init=` and `max_tasks_per_worker=`
- **Interpreter executor** (Python 3.14+) — `executor="interpreter"`: true CPU parallelism for pure-Python work without process overhead (PEP 734)
- **Contextvars propagation** — correlation IDs survive into thread workers
- **Decorator API** — `@parallel` / `@async_parallel` with `.map()`, `.starmap()`, `.stream()` — typed options via `Unpack[TypedDict]`, and single-arg named functions bind their item type (`fetch.map([1])` is a type error when `fetch` takes `str` — in mypy *and* pyright)

## Quick Start

### Sync

```python
import requests
from pyarallel import parallel_map, RateLimit, Retry

def fetch(url):
    return requests.get(url, timeout=10).json()

# Fan out over a list, get ordered results
result = parallel_map(fetch, urls, workers=10)

# Rate-limited API calls with retry
def call_api(user_id):
    return requests.get(f"https://api.example.com/users/{user_id}").json()

result = parallel_map(
    call_api, user_ids,
    workers=10,
    rate_limit=RateLimit(100, "minute"),
    retry=Retry(attempts=3, backoff=1.0, on=(ConnectionError, TimeoutError)),
)

# CPU-bound with processes
from PIL import Image

def resize_image(path):
    img = Image.open(path)
    img.thumbnail((800, 600))
    img.save(path.replace(".png", "_thumb.png"))

result = parallel_map(resize_image, paths, executor="process")
```

### Async

```python
import httpx
from pyarallel import async_parallel_map

client = httpx.AsyncClient()  # ONE client: pooled connections across all calls

async def fetch_async(url):
    return (await client.get(url, timeout=10)).json()

result = await async_parallel_map(
    fetch_async, urls, concurrency=20, task_timeout=5.0,
)
await client.aclose()
```

### Decorator

Adds `.map()`, `.starmap()`, `.stream()` without changing the function:

```python
from pyarallel import parallel, async_parallel, RateLimit

@parallel(workers=8, rate_limit=RateLimit(100, "minute"))
def fetch(url):
    return requests.get(url).json()

fetch("http://example.com")          # normal call — returns dict
fetch.map(urls)                      # parallel — returns ParallelResult
fetch.stream(urls, window_size=500)   # streaming — yields ItemResult

client = httpx.AsyncClient()  # ONE client, reused by every call

@async_parallel(concurrency=10)
async def fetch_async(url):
    return (await client.get(url)).json()

await fetch_async.map(urls)          # async parallel
```

### Streaming — Constant Memory

For ETL, pipelines, or datasets too large to hold in memory:

```python
from pyarallel import parallel_iter

def transform(row):
    return {"id": row["id"], "name": row["name"].strip().title()}

for item in parallel_iter(transform, ten_million_rows, window_size=1000):
    if item.ok:
        db.save(item.value)
    else:
        log_error(item.index, item.error)
```

### Error Handling

All errors collected, never silently swallowed:

```python
def send_email(msg):
    return smtp.send(msg["to"], msg["subject"], msg["body"])

result = parallel_map(send_email, messages)

if result.ok:
    values = result.values()           # list of all results, in order
else:
    for idx, exc in result.failures():
        log_error(idx, exc)
    result.raise_on_failure()          # or raise ExceptionGroup with all errors
```

## API Summary

| Function | Decorator | Returns | Use case |
|---|---|---|---|
| `parallel_map(fn, items)` | `.map(items)` | `ParallelResult` | Results fit in memory |
| `parallel_starmap(fn, items)` | `.starmap(items)` | `ParallelResult` | Multi-arg, fits in memory |
| `parallel_iter(fn, items)` | `.stream(items)` | `Iterator[ItemResult]` | Streaming, constant memory |

Async mirrors: `async_parallel_map`, `async_parallel_starmap`, `async_parallel_iter`

| Config | Example |
|---|---|
| `RateLimit(count, per, burst)` | `RateLimit(100, "minute", burst=20)` |
| `Limiter(rate_limit)` | shared budget: `Limiter(RateLimit(100, "minute"))` |
| `Retry(attempts, backoff, on, retry_if, wait_from)` | `Retry(attempts=3, on=(ConnectionError,))` |
| `Retry.for_http(on, statuses)` | HTTP prewired: `Retry.for_http(on=(httpx.HTTPStatusError,))` |
| `checkpoint=` | resumable runs: `checkpoint="run.ckpt"` |

Works with instance methods and static methods via `@parallel` decorator — see [full docs](https://oneryalcin.github.io/pyarallel/).

## See It Prove Itself

Every *resilience* claim — 429 handling, the pool-wide pause, rate-limit
pacing, kill-and-resume — verified locally in ~10 seconds, no
credentials, no dependencies, one file:

```bash
python examples/resilience_demo.py
```

A fake quota-enforcing API comes up on localhost; a full-speed pool
draws a 429 and you watch **one** `Retry-After` pause the whole pool
(the server measures the gap); the same job with a client-side
`RateLimit` never gets throttled at all; then a checkpointed run is
**SIGKILLed mid-flight** and rerun — it resumes from SQLite, and the
server's request counter proves the paid-for calls were not repeated.
The demo asserts its own claims and exits non-zero if any fail.

## Documentation

[Full docs](https://oneryalcin.github.io/pyarallel/) — API reference, advanced features, best practices.

Deciding whether pyarallel fits? [Comparison vs tenacity+ThreadPoolExecutor, aiometer, mpire, joblib](https://oneryalcin.github.io/pyarallel/getting-started/comparison/) — including when *not* to use it. Then the [Cookbook](https://oneryalcin.github.io/pyarallel/cookbook/): batch LLM calls, embeddings with resume, polite scraping, bulk GitHub/registry ops, NCBI fetches, secrets rotation.

## License

MIT — see [LICENSE.md](LICENSE.md).
