pyarallel 0.9.0

The fan-out layer for rate-limited APIs — parallel map with rate limiting, retry, and structured errors. Sync and async, zero dependencies.

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:

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:

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:

import httpx
from pyarallel import async_parallel_map, RateLimit, Retry

async def fetch_async(url):
    async with httpx.AsyncClient() as client:
        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)),
)
# Same result model — result.ok, result.successes(), result.failures()

Install

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

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

import httpx
from pyarallel import async_parallel_map

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

result = await async_parallel_map(
    fetch_async, urls, concurrency=20, task_timeout=5.0,
)

Decorator

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

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

@async_parallel(concurrency=10)
async def fetch_async(url):
    async with httpx.AsyncClient() as c:
        return (await c.get(url)).json()

await fetch_async.map(urls)          # async parallel

Streaming — Constant Memory

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

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:

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.

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:

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 — API reference, advanced features, best practices.

Deciding whether pyarallel fits? Comparison vs tenacity+ThreadPoolExecutor, aiometer, mpire, joblib — including when not to use it. Then the Cookbook: batch LLM calls, embeddings with resume, polite scraping, bulk GitHub/registry ops, NCBI fetches, secrets rotation.

License

MIT — see LICENSE.md.