DEV Community

Aritro Saha
Aritro Saha

Posted on

Your overnight script died at item 5,247. Again. I built a decorator so it never does.

You know this loop. You've written it a hundred times:

for item in items:
    process(item)
Enter fullscreen mode Exit fullscreen mode

You kick it off against 10,000 records — CSV rows, URLs to scrape, images to resize — and go to bed. In the morning you find it crashed at item #5,247 on something malformed you never anticipated. 5,246 items of finished work, gone. Three hours, gone.

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

I got tired of this cycle, so I built quarantine. It's one decorator:

from quarantine import quarantine

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

for item in items:
    process(item)
Enter fullscreen mode Exit fullscreen mode

That's the whole integration. Bad items no longer kill your job. Each failure gets saved to a .quarantine/ folder — with its full traceback and the exact input that caused it — and your loop keeps running. At the end:

✓ 9,996 processed · ✗ 4 quarantined → .quarantine/  (run `quarantine retry` after fixing)
Enter fullscreen mode Exit fullscreen mode
pip install quarantine-py
Enter fullscreen mode Exit fullscreen mode

The whole workflow in 25 seconds: the run survives 4 bad rows, quarantine list shows them, quarantine retry recovers them

"Can't I just use try/except?"

Yes! In fact, try/except is exactly what quarantine uses under the hood. But here's what the hand-rolled version looks like 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?
Enter fullscreen mode Exit fullscreen mode

Every ❌ is a real problem you'd end up solving yourself, in every script, forever:

try/except by hand @quarantine
Loop survives bad items
Failures survive a crash/restart ❌ in RAM, gone ✅ saved to disk instantly
Full traceback kept for later ❌ usually just printed ✅ stored with the item
The exact bad input saved ❌ serialize it yourself ✅ automatic
Re-run only the failures ❌ build it yourself quarantine retry
Debug with the real bad input ❌ archaeology in logs quarantine debug 2
Detects "everything is failing, stop" ✅ halts on failure streaks
Skips already-known-bad items on rerun

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

If you've worked with message queues, you'll recognize the pattern: it's a dead-letter queue, ported from broker infrastructure to a plain Python for-loop. No broker, no server, no config. One decorator.

The workflow

1. See what failed:

$ 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, ...}
Enter fullscreen mode Exit fullscreen mode

2. Fix your code, then retry only the failures:

$ quarantine retry
✓ 3 recovered · ✗ 1 still failing (kept in quarantine)
Enter fullscreen mode Exit fullscreen mode

No rerunning the 9,996 items that already worked.

3. Debug with the actual bad input:

$ quarantine debug 2
# opens pdb on the frame that raised, with the exact input that failed
Enter fullscreen mode Exit fullscreen mode

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

There's also quarantine stats, quarantine show, --json output for piping, and even quarantine ui for a local web dashboard.

The safety valve

Here's a subtle failure mode of "just keep going": if 50 items fail in a row, that's not bad data — that's your database being down. Quarantining all 10,000 items would be silly. So quarantine ships a circuit breaker:

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

Tune it with @quarantine(halt_after=100).

No magic — just files

The .quarantine/ folder is plain files you can inspect yourself:

.quarantine/
├── 0001/
│   ├── input.pkl        # the exact item (pickle, JSON fallback)
│   ├── input.txt        # human-readable repr
│   ├── traceback.txt    # full error, exactly as it would have printed
│   └── meta.json        # function, timestamp, attempt count, versions
└── index.json
Enter fullscreen mode Exit fullscreen mode

Some design decisions I care about:

  • Atomic writes — a crash mid-save never corrupts the folder.
  • Redaction before diskredact=["api_key", "password"] means secrets never touch the filesystem.
  • Serialization fallbacks — pickle → JSON → repr. Something readable is always saved, even for exotic objects.
  • Dedup on rerun — items already in quarantine are skipped instead of spamming duplicates.

It scales down and up

For quick scripts, the decorator with zero arguments is all you need. When you want more:

@quarantine(
    only=(ValueError, KeyError),   # only quarantine these; others still crash
    retries=2, backoff=0.5,        # retry transient failures with backoff
    halt_after=50,                 # the circuit breaker
    redact=["api_key"],            # scrub secrets before saving
    on_quarantine=slack_alert,     # hooks for alerting/metrics
    dir="s3://bucket/prefix",      # one shared quarantine for a fleet of workers
)
def process(item): ...
Enter fullscreen mode Exit fullscreen mode

It works on async def, it's safe under threads and multiprocessing (proven with real concurrent processes in the test suite), and there's a shield() helper if you'd rather wrap a loop than decorate a function:

from quarantine import shield

for result in shield(items, using=process):
    ...
Enter fullscreen mode Exit fullscreen mode

The stats that matter:

  • Zero runtime dependencies — standard library only (S3 backend optionally adds boto3)
  • Fully typed, ships py.typed, passes strict mypy
  • Python 3.10+, tested on Linux/macOS/Windows in CI
  • MIT licensed

When you should NOT use it

Honesty section:

  • You want the crash. In a bank transfer pipeline, stopping on the first error might be correct. Continuing is a choice — make it deliberately.
  • You're already on Celery/Kafka/Airflow. Those have real dead-letter queues; use them. quarantine is for the 95% of scripts that will never justify that machinery.
  • Failures are expected and normal ("404 means skip"). Handle those with a normal if — quarantine is for unexpected failures you'll want to investigate.

Try it

pip install quarantine-py
Enter fullscreen mode Exit fullscreen mode

If your overnight job has ever died at item 5,247 — this one's for you. I'd love to hear what you think, and issues/PRs are very welcome. What's next on the roadmap: more storage backends (GCS, Azure, Redis, Databricks).

Top comments (0)