When one bad item crashes your loop of 10,000 — don't crash. Set it aside, keep going, fix it later.

pip install quarantine-py

from quarantine import quarantine

@quarantine def process(item): ... # your normal code, unchanged

for item in items: process(item)

That's it. Bad items no longer kill your job. They get saved to a .quarantine/ folder — with their full error and the exact input that caused it — and your loop keeps running.

✓ 9,996 processed · ✗ 4 quarantined → .quarantine/ (run quarantine retry after fixing)

The problem (explained like you're new to this)

Imagine you're processing 10,000 records — rows from a CSV, URLs to scrape, images to resize. You write a loop:

for item in items: process(item)

You run it. It works... until item #5,247, which is malformed in some way you didn't expect. Your script crashes. You lost 5,246 items of finished work and 3 hours.

So you fix the bug for that one weird item and rerun from the start. Three hours later it crashes again — at item #7,913, for a different reason.

This cycle is one of the most common, most painful experiences in programming. quarantine ends it.

Yes — try/except is exactly what quarantine uses under the hood. But here's the try/except version once you make it actually safe:

failed = [] for item in items: try: process(item) except Exception as e: failed.append(item) # ❌ lost forever if the script dies later print(f"failed: {e}") # ❌ traceback gone — good luck debugging tomorrow

❌ how do I re-run JUST these failures after I fix the bug?

❌ what if 500 fail in a row because the API is down — keep going?!

❌ how do I save a weird object (DataFrame row? bytes?) to look at later?

Every ❌ is a real problem you'd have to solve yourself, in every script, forever. quarantine solves them once:

quarantine is not a replacement for try/except. It's the 200 lines of bookkeeping you'd have to write around it — done correctly, once.

Hospitals don't shut down when one patient has an infection. They quarantine the patient, treat everyone else, and come back with the right medicine.

items ──▶ process() ──▶ ✓ done │ ✗ raises an exception ▼ .quarantine/ ◀── the "sick bay" folder ├── item + its data ├── the full error traceback └── when/why it failed

Your job finishes. The sick items wait for you, with their full medical charts.

Available on PyPI and conda-forge as quarantine-py:

pip install quarantine-py

conda / mamba / pixi

From conda-forge:

conda

conda install -c conda-forge quarantine-py

mamba

mamba install -c conda-forge quarantine-py

pixi

pixi add quarantine-py

uv

uv add quarantine-py # inside a uv project uv pip install quarantine-py # standalone with uv

pipx (CLI standalone)

If you only want the standalone quarantine command-line tool isolated:

pipx install quarantine-py

For development / from source

Install latest unreleased main directly from GitHub

pip install git+https://github.com/halcyon-past/quarantine

Or clone for local development

git clone https://github.com/halcyon-past/quarantine.git cd quarantine pip install -e ".[dev]" # includes pytest, ruff, mypy, hypothesis

Requirements

Installing also puts a quarantine command on your PATH. Check both halves:

$ quarantine --version quarantine 1.2.0 $ python -c "import quarantine; print(quarantine.version)" 1.2.0

If the command is not found (a common --user install wrinkle), the module form always works and takes the same arguments:

python -m quarantine list

Nothing needs configuring. The first time a call fails, .quarantine/ appears next to wherever you started Python. Add it to your .gitignore - quarantined inputs are real data, and real data does not belong in git.

.quarantine/

Usage

1. Basic — decorate and forget

from quarantine import quarantine

@quarantine def process(row): price = float(row["price"]) # crashes on "N/A"? quarantined. save_to_db(row["id"], price)

for row in rows: process(row)

2. See what got quarantined

$ quarantine list

when function error input preview

1 09:14:02 process ValueError: could not convert {'id': 8812, 'price': 'N/A', ...} 2 09:31:44 process KeyError: 'price' {'id': 9107, ...}

3. Fix your code, then retry only the failures

$ quarantine retry ✓ 3 recovered · ✗ 1 still failing (kept in quarantine)

No rerunning the 9,996 items that already worked.

(Note: If your function lives in a standalone script rather than an installed package, use -i to tell quarantine where to import it from: quarantine retry -i my_script.py)

$ quarantine debug 2

opens a Python debugger with item set to the exact input that failed

(You can use -i here too: quarantine debug 2 -i my_script.py)

The single biggest time-saver: you never have to reproduce the bug. The bug's exact input is sitting on disk.

If 50 items fail in a row, that's not bad data — that's your database being down. Quarantining 10,000 items would be silly. quarantine halts instead:

✋ 50 consecutive failures — this looks systemic, not bad data. Halting. Last error: ConnectionError: db.internal:5432 refused

Tune it: @quarantine(halt_after=100).

@quarantine( dir=".quarantine", # where the sick bay lives - a folder, or s3://bucket/prefix only=(ValueError, KeyError), # only quarantine these; others still crash halt_after=50, # consecutive-failure circuit breaker max_items=10_000, # cap disk usage retries=2, # retry transient failures backoff=0.5, # base delay between retries backoff_factor=2.0, # grow the delay per retry (exponential backoff) jitter=0.25, # randomise the delay so workers spread out dead_after=5, # records failing 5+ retries are skipped by blanket retries redact=["api_key", "password"], # scrub these fields before saving inputs on_quarantine=my_alert_fn, # e.g., send a Slack ping on_retry_success=my_metric_fn, # hooks for the retry lifecycle too ) def process(item): ...

7. Works on loops too, without a decorator

from quarantine import shield

for item in shield(items, using=process): ...

What's in the .quarantine/ folder?

Plain files. No database, no magic — you can inspect everything yourself:

.quarantine/ ├── 0001/ │ ├── input.pkl # the exact item (pickle, JSON fallback for simple data) │ ├── input.txt # human-readable repr, so you can just look at it │ ├── traceback.txt # full error, exactly as it would have printed │ └── meta.json # function name, timestamp, attempt count, python/pkg versions └── index.json

Design rules:

Every command takes -d/--dir PATH (default: $QUARANTINE_DIR, else ./.quarantine), and list, show, retry and stats all take --json so you can pipe them somewhere useful.

fail a nightly job if anything is still sitting in quarantine

quarantine retry || echo "still broken - look at: $(quarantine list -n 3)"

Python API reference

from quarantine import quarantine, shield, Quarantine, QUARANTINED, records, retry

Decorating and looping

Return values. A quarantined call returns the QUARANTINED sentinel; an input recognised as already-bad returns SKIPPED. Both are falsy, so if process(item): does the sensible thing. Use is_quarantined(result) / is_skipped(result) when you want to be explicit.

The explicit object, when you would rather pass something around than rely on a decorator:

from quarantine import Quarantine

q = Quarantine("build/bad-rows", halt_after=10, redact=["api_key"])

safe = q.wrap(process) # same as the decorator q.call(process, item) # one-off call, same protection await q.acall(fetch, url) # async one-off q.records() # list[Record], oldest first q.retry() # -> RetryResult(recovered, still_failing, unretryable) await q.aretry() # for records from async functions q.clear() # empty the folder q.stats # Stats(processed, quarantined, skipped, recovered) q.summary_line() # the one-line report, or None len(q), list(q) # how many records; iterate them

Module-level shortcuts operate on the default folder (or dir=): records(), retry(), aretry(), clear(), summary().

A Record is what you get back from records():

record = records()[0] record.id # 1 record.function # "process" (qualified name in .qualified_name) record.error_type # "ValueError" record.summary # "ValueError: could not convert string to float: 'N/A'" record.attempts # 2, after one retry record.redacted # ["api_key"] - what was scrubbed record.path # Path(".quarantine/0001") record.traceback_text() # the stored traceback record.load_call() # Call(args=({...},), kwargs={}) - the original input record.load_call().item # the item itself

Exceptions (all subclass QuarantineError, and none of them are ever quarantined themselves):

Rendered and searchable at quarantine-py.aritro.cloud, or browse the sources:

Is this just a dead-letter queue? Yes — that's exactly the pattern, ported from message-queue infrastructure to a plain Python for-loop. No broker, no server, one decorator.

Async? @quarantine works on async def too.

Threads/processes? Yes — safe under threading, multiprocessing and concurrent.futures. Writes are atomic, ids cannot collide, and the folder is append-only per item; the test suite proves it with real concurrent processes sharing one folder. Details in docs/usage.md.

Made for everyone whose overnight job died at item 5,247.