Skip to main content

Concurrency and state

A task can do many things at once, run more than once, and carry state between runs. This page covers running work in parallel with aisle.parallel, what happens when a run repeats, and the SDK tools — aisle.checkpoint, aisle.cache, and aisle.run — that let a repeat pick up where the last one left off instead of redoing finished work.

Running in parallel​

aisle.parallel runs a function over a list of items concurrently. Instead of a plain loop that calls a function once per item in sequence, you hand it the function and the items and it fans the calls out across a pool of workers, in one call.

Reach for it when you have many items that each do I/O — an API call, a model call, a database read — and running them one at a time would be slow. For a handful of items, or work that is pure computation, a plain loop is simpler.

The call​

def enrich(account):
    person = aisle.integrations.pipedrive.get_person(id=account["id"])
    return {"id": account["id"], "email": person["email"]}

result = aisle.parallel(enrich, accounts, concurrency=10)

for row in result.results:
    print(row)

The function comes first, then the items. aisle.parallel(fn, items) calls fn(item) for each item using a pool of worker threads and returns a result object:

  • result.results — each call's return value, in input order (not completion order).
  • result.completed — how many items ran.
  • result.skipped — how many were skipped via a checkpoint (see below).
  • result.errors — collected errors, when continue_on_error=True.

Concurrency​

concurrency sets how many workers run at once (default 5). Higher means more calls in flight, which is faster until you hit the downstream service's limits.

result = aisle.parallel(enrich, accounts, concurrency=20)

Rate limiting​

max_per_minute caps the call rate across all workers, so a fast pool doesn't exceed a provider's per-minute quota. It is a shared leaky bucket — you don't hand-roll a backoff.

result = aisle.parallel(
    enrich,
    accounts,
    concurrency=20,
    max_per_minute=100,
)

Twenty workers still run, but no more than 100 calls go out in any minute.

Retries​

retry retries a failed fn(item) call. It accepts a few forms:

  • retry=3 — up to 3 attempts (2 retries) per item.
  • retry=True — the default policy (3 attempts).
  • retry=False or omitted — no retry, a single call.
  • retry=aisle.RetryPolicy(...) — full control.
result = aisle.parallel(enrich, accounts, concurrency=10, retry=3)

Retries use full-jitter exponential backoff. Errors that indicate a bug rather than a transient failure are not retried: ValueError, TypeError, KeyError, AttributeError, IndexError, NotImplementedError, and AssertionError. Retrying those would just fail again.

aisle.RetryPolicy tunes the behaviour:

policy = aisle.RetryPolicy(
    max_attempts=5,
    initial_backoff_ms=500,
    max_backoff_ms=30_000,
    backoff_multiplier=2.0,
    non_retryable=(MyFatalError,),
)

result = aisle.parallel(enrich, accounts, retry=policy)
  • max_attempts — total tries including the first (1 = no retry).
  • initial_backoff_ms, max_backoff_ms, backoff_multiplier — the backoff curve.
  • non_retryable — extra exception types to treat as fatal, added to the defaults above.
  • non_retryable_predicate — a callable fn(exc) -> bool for cases the type alone can't decide, such as an HTTP 4xx versus 5xx that share one exception class. Return True to skip the retry.
def is_client_error(exc):
    return getattr(exc, "status_code", 500) < 500

policy = aisle.RetryPolicy(
    max_attempts=4,
    non_retryable_predicate=is_client_error,
)

Continuing past errors​

By default the first unhandled error stops the whole run and raises. Set continue_on_error=True to keep going and collect failures instead, then inspect result.errors:

result = aisle.parallel(enrich, accounts, continue_on_error=True)

print(f"{result.completed} enriched, {len(result.errors)} failed")
for failure in result.errors:
    print(failure["item"], "→", failure["error"])

Each entry in result.errors is a dict with the offending item and the error string.

Skipping finished work with a checkpoint​

checkpoint= names a store that records which items finished, so a later run — an auto-retry, or a resumed long run — skips the ones already done instead of repeating them:

result = aisle.parallel(
    process_invoice,
    invoices,
    concurrency=8,
    checkpoint="invoices",
    checkpoint_key="id",
    retry=3,
)
  • checkpoint="invoices" — the namespace to record progress under.
  • checkpoint_key — how to identify an item. By default it's the item's position in the list; pass a string to read a field ("id" → item["id"]), or a callable for anything else (checkpoint_key=lambda item: item["ref"]).
  • store_result=True — also store each item's return value, so a resumed run returns the stored value in result.results rather than re-running the item.

A retried run then re-processes only the invoices that hadn't finished. This is aisle.parallel's wrapper over the same aisle.checkpoint store described below.

When a call runs more than once​

Because a retry or a checkpoint resume can call fn(item) again, its side effects must be safe to repeat. Make writes idempotent — INSERT ... ON CONFLICT, an upsert, a "post only if not already posted" guard — so a second call doesn't double-write. Aisle does not roll back partial state from a failed call.

What a rerun is​

A task can run more than once — you run it again by hand, or Aisle retries it after a failure. When a run repeats, you usually don't want it to redo work that already succeeded: re-post a message, re-charge a model call, re-process the same records.

  • Manual re-run — click Run Now again. This is a fresh run with its own state.
  • Retry a finished run — a run that errored, timed out, or was cancelled shows a Retry button on its execution page. The retry re-runs the script.
  • Auto-retry on failure — turn it on in Settings under Advanced. A failed run is re-run automatically, up to the max retries and after the backoff you set.

An auto-retry and a manual Retry both re-run the whole script from the top. The tools below let a re-run skip finished work.

aisle.checkpoint — skip finished work across a retry​

aisle.checkpoint is a key/value store scoped to a single run and the retries of that run. A retry sees the keys the earlier attempt wrote; a fresh, unrelated run does not. Use it to guard steps so a retry doesn't repeat their side effects.

  • aisle.checkpoint.mark(key) — record that a step finished (no stored value).
  • aisle.checkpoint.done(key) — True if the key was marked or set.
  • aisle.checkpoint.set(key, value) — store a value (also marks the key done).
  • aisle.checkpoint.get(key) — the stored value, or None if absent or mark-only.
  • aisle.checkpoint.get_or_set(key, fn) — return the stored value, or call fn(), store it, and return it.
  • aisle.checkpoint.query_prefix(prefix) — every key/value pair whose key starts with prefix.

Guard a side effect with mark/done:

for ticket in tickets:
    key = f"notified:{ticket['id']}"
    if aisle.checkpoint.done(key):
        continue

    aisle.integrations.slack.create_message(
        channel="#alerts",
        text=f"New ticket: {ticket['subject']}",
    )
    aisle.checkpoint.mark(key)

If the run fails partway through and retries, the tickets already notified are skipped — no duplicate Slack messages.

Use get_or_set when the step produces a value you want to keep, like an expensive model call:

summary = aisle.checkpoint.get_or_set(
    f"summary:{doc['id']}",
    lambda: aisle.ai.raw(f"Summarize:\n\n{doc['text']}"),
)

The model runs once. A retry reuses the stored summary instead of paying for it again.

get returns None for a key written with mark (which stores no value). If a step can legitimately produce None, use explicit set/get rather than get_or_set.

For fan-out work, aisle.parallel has this built in — pass checkpoint= and a retried run skips the items that already finished. See Running in parallel above.

aisle.cache — reuse expensive results across runs​

aisle.checkpoint is per-run. aisle.cache is per-task: it's shared across every run of the task, and each value carries a TTL that says how long it stays valid.

  • aisle.cache.get(key) — the value, or None if absent or expired.
  • aisle.cache.set(key, value, ttl) — store a value with a TTL.
  • aisle.cache.get_or_set(key, fn, ttl) — return the cached value, or call fn(), store it with the TTL, and return it.
  • aisle.cache.delete(key) — evict a value.

ttl is an integer number of seconds, or a string like "30s", "6h", or "7d".

rates = aisle.cache.get_or_set(
    "fx-rates",
    lambda: aisle.http.request("GET", "https://api.example.com/rates").json(),
    ttl="6h",
)

Within a run this avoids fetching the same thing twice. Across runs it holds a result that's still good later, so a nightly job doesn't re-pull reference data that hasn't changed. Reach for cache for reference data and expensive lookups you're willing to reuse for a while; reach for checkpoint to make a single run's retry idempotent.

aisle.cache needs a saved task to namespace its entries. On an unsaved draft run it raises a clear error — save the task first.

aisle.run.last_run_at — process only what's new​

aisle.run.last_run_at is the time this task last completed successfully, or None on the first run. Use it to process only what changed since then rather than everything every time:

since = aisle.run.last_run_at or aisle.dates.parse("2024-01-01T00:00:00Z")

orders = aisle.integrations.shopify.list_orders(
    updated_after=since.isoformat(),
)

for order in orders:
    handle(order)

Because it tracks the last successful run, a run that fails doesn't advance the watermark — the next run reprocesses the same window instead of skipping it.

aisle.run.is_test — stay quiet in a draft​

A draft run from the editor sets aisle.run.is_test to True. Branch on it so a test doesn't fire real side effects while you're iterating:

if aisle.run.is_test:
    print(f"[draft] would notify {len(recipients)} recipients")
else:
    for recipient in recipients:
        aisle.integrations.slack.create_message(channel=recipient, text=body)

Where to go next​

Last updated on