Scraper API Retry Handling for Production Pipelines

export EVOMI_API_KEY=your-api-key
export EVOMI_CONCURRENCY=5  # your slot count

# /// script
# requires-python = ">=3.11"
# dependencies = ["httpx>=0.28,<1"]
# ///
import asyncio
import hashlib
import os
import pathlib
import random
import re
import sys

import httpx

BASE = "https://scrape.evomi.com/api/v1/scraper"
if not os.environ.get("EVOMI_API_KEY"):
    raise SystemExit("Set EVOMI_API_KEY to the key from your Evomi dashboard")
HEADERS = {"x-api-key": os.environ["EVOMI_API_KEY"]}
# Your plan's slot count: every accepted response shows X-RateLimit-Limit
CONCURRENCY = int(os.environ.get("EVOMI_CONCURRENCY", "5"))
CREDIT_BUDGET = 12  # No attempt starts if it could push a URL past this
TASK_DEADLINE = 300  # Seconds to wait for one task, then no_result
COST = {"request": 2, "browser": 5}  # Credits per call, residential proxies
# Append {"mode": "browser", "proxy_country": "GB"} to try another region
STEPS = [{"mode": "request"}, {"mode": "browser"}]
STOP_BATCH = {401, 402}  # Bad key or no credits: every URL would fail
BLOCK_TITLES = (
    "robot or human",
    "just a moment",
    "access denied",
    "attention required",
    "access to this page has been denied",
)


class StopBatch(Exception):
    """The key or the credits need attention before anything else runs."""


class Rejected(Exception):
    """The API refused this request's parameters. Fix them, don't retry."""


class TaskLost(Exception):
    """No task record exists for this task ID."""


def backoff(attempt, base=1.0, cap=20.0):
    # Full jitter: a random wait, so workers don't retry in the same second
    return random.uniform(0, min(cap, base * 2**attempt))


async def submit(client, body):
    """Send the paid POST. Resend only when the API rejected it."""
    for n in range(10):
        try:
            r = await client.post(f"{BASE}/realtime", headers=HEADERS, json=body)
        except (httpx.ConnectError, httpx.ConnectTimeout):
            if n == 9:
                raise
            await asyncio.sleep(backoff(n))  # The request was never sent
            continue
        if r.status_code not in (429, 503):
            break
        await asyncio.sleep(backoff(n))  # Rejected, so no task was created
    if r.status_code in STOP_BATCH:
        raise StopBatch(f"{r.status_code}: {r.text[:200]}")
    if r.status_code in (400, 422):
        raise Rejected(r.status_code)
    r.raise_for_status()
    return r.json()["task_id"]


async def wait_for(client, task_id):
    """Poll the task. A repeated GET never creates a new task."""
    async with asyncio.timeout(TASK_DEADLINE):
        while True:
            await asyncio.sleep(2)
            try:
                r = await client.get(f"{BASE}/tasks/{task_id}", headers=HEADERS)
            except httpx.TransportError:
                continue
            if r.headers.get("x-task-finished-at"):
                return r.json()  # Finished: a 500 here is the task's result
            if r.status_code == 404:
                raise TaskLost(task_id)
            if r.status_code in STOP_BATCH:
                raise StopBatch(f"{r.status_code} while reading a task")
            # A 202 while running, or a temporary error: read again


def has_text(html, minimum=500):
    """Default check: enough visible text outside scripts and styles."""
    text = re.sub(r"<(script|style)\b.*?</\1>|<[^>]+>", " ", html, flags=re.I | re.S)
    return len(" ".join(text.split())) >= minimum


def save(url, html):
    """Write a usable page to pages/ and list it in pages/index.tsv."""
    folder = pathlib.Path("pages")
    folder.mkdir(exist_ok=True)
    path = folder / f"{hashlib.sha256(url.encode()).hexdigest()[:16]}.html"
    path.write_text(html, encoding="utf-8")
    with open(folder / "index.tsv", "a", encoding="utf-8") as index:
        index.write(f"{path.name}\t{url}\n")
    return path


def target_headers(job):
    raw = (job.get("headers") or {}).get("response_headers") or {}
    return {k.lower(): v for k, v in raw.items()}


def classify(job, looks_ok):
    code, html = job.get("status_code"), job.get("content") or ""
    if not job.get("success"):
        return "api_error"
    if code in (404, 410):
        return "gone"
    if code == 429:
        return "throttled"  # The site is rate-limiting you
    title = re.search(r"<title[^>]*>(.*?)</title>", html, re.I | re.S)
    title = title.group(1).strip().lower() if title else ""
    challenged = target_headers(job).get("cf-mitigated") == "challenge"
    if code == 403 or challenged or title.startswith(BLOCK_TITLES):
        return "blocked"
    if code and code >= 500:
        return "target_error"
    if not code or not 200 <= code < 300:
        return "unexpected"  # Review it before you write a rule for it
    return "ok" if looks_ok(html) else "invalid"


async def scrape(client, slots, url, looks_ok):
    spent, attempts, outcome, code = 0.0, 0, "failed", None
    for step in STEPS:
        for retry in range(3):
            if spent + COST[step["mode"]] > CREDIT_BUDGET:
                return url, "over_budget", code, attempts, spent
            # No proxy_session_id: each call gets a rotating exit IP
            body = {
                "url": url,
                **step,
                "async": True,
                "delivery": "json",
                "include_content": True,
                "capture_headers": True,
            }
            async with slots:  # A slot stays busy until the task ends
                task_id = await submit(client, body)
                # The service saves (url, task_id) to resume after a crash
                try:
                    job = await wait_for(client, task_id)
                except (TimeoutError, TaskLost) as exc:
                    outcome = f"no_result ({type(exc).__name__})"
                    return url, outcome, code, attempts + 1, spent
            attempts += 1
            spent += float(job.get("credits_used") or 0)
            code = job.get("status_code")
            outcome = classify(job, looks_ok)
            if outcome == "ok":
                save(url, job["content"])
            if outcome in ("ok", "gone"):
                outcome = f"{outcome} via {job.get('mode_used')}"
                return url, outcome, code, attempts, spent
            if outcome == "throttled":  # Return Retry-After for a later retry
                wait = target_headers(job).get("retry-after")
                return (url, f"throttled, retry-after={wait}", code, attempts, spent)
            if outcome == "unexpected":
                return url, outcome, code, attempts, spent
            if outcome in ("api_error", "target_error"):
                if retry == 2:  # A different mode will not fix it
                    return url, outcome, code, attempts, spent
                await asyncio.sleep(backoff(retry))
                continue  # Transient: same step again after a pause
            break  # Blocked or invalid: this step returns the same page
    return url, outcome, code, attempts, spent


async def run(client, slots, url, looks_ok):
    try:
        result = await scrape(client, slots, url, looks_ok)
    except StopBatch:
        raise  # The TaskGroup cancels every other URL
    except Rejected as exc:  # Includes a domain that doesn't exist
        result = url, f"rejected ({exc})", None, 0, 0.0
    except Exception as exc:
        result = url, f"error ({type(exc).__name__})", None, 0, 0.0
    url, outcome, code, attempts, spent = result
    # Print as each URL finishes, so a long batch shows progress
    print(
        f"{outcome:<28} site={code!s:<4} attempts={attempts}  "
        f"credits={spent:g}  {url}",
        flush=True,
    )
    return result


async def main(urls, looks_ok=has_text):
    slots = asyncio.Semaphore(CONCURRENCY)
    timeout = httpx.Timeout(30.0, connect=10.0)  # Async calls return fast
    stopped = None
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            async with asyncio.TaskGroup() as group:
                tasks = [
                    group.create_task(run(client, slots, u, looks_ok)) for u in urls
                ]
    except* StopBatch as stop:
        stopped = stop.exceptions[0]
    if stopped:  # Exit with an error, so a scheduler sees the failure
        raise SystemExit(f"Batch stopped: {stopped}")
    total = sum(task.result()[4] for task in tasks)
    print(f"done: {len(tasks)} URLs, {total:g} credits")


if __name__ == "__main__":
    # Pass a file with one URL per line, or run the sample list below
    urls = [
        "https://www.walmart.com/search?q=coffee+maker",
        "https://quotes.toscrape.com/js/",
        "https://www.python.org/this-page-does-not-exist",
        "https://httpbin.io/status/503",
        "https://httpbin.io/status/429",
    ]
    if len(sys.argv) > 1:
        path = pathlib.Path(sys.argv[1])
        if not path.is_file():
            raise SystemExit(f"No URL file at {path}")
        urls = path.read_text().split()
    asyncio.run(main(urls))

uv run scrape_with_retries.py

gone via request             site=404  attempts=1  credits=2  https://www.python.org/this-page-does-not-exist
throttled, retry-after=None  site=429  attempts=1  credits=2  https://httpbin.io/status/429
ok via browser               site=200  attempts=2  credits=7  https://quotes.toscrape.com/js/
target_error                 site=503  attempts=3  credits=6  https://httpbin.io/status/503
ok via browser               site=200  attempts=2  credits=7  https://www.walmart.com/search?q=coffee+maker
done: 5 URLs, 24 credits

# /// script
# requires-python = ">=3.11"
# dependencies = ["httpx>=0.28,<1", "aiohttp>=3.14,<4"]
# ///
import asyncio
import email.utils
import hashlib
import hmac
import json
import os
import sqlite3
import time
from urllib.parse import urlsplit

import httpx
from aiohttp import web

from scrape_with_retries import (
    BASE,
    CONCURRENCY,
    COST,
    HEADERS,
    STEPS,
    backoff,
    classify,
    has_text,
    save,
    target_headers,
)

# Leave EVOMI_WEBHOOK_URL unset to run locally: the poller reads tasks
HOOK_URL = os.environ.get("EVOMI_WEBHOOK_URL")  # Public HTTPS endpoint
SECRET = os.environ.get("EVOMI_WEBHOOK_SECRET", "")  # Long and random
if HOOK_URL and not SECRET:
    raise SystemExit("Set EVOMI_WEBHOOK_SECRET to sign webhook deliveries")
PORT = int(os.environ.get("PORT", "8080"))
CREDIT_BUDGET, MAX_RETRIES = 12, 2
PAUSE_AFTER_BLOCKS, PAUSE_SECONDS, LEARN_SECONDS = 3, 3600, 86400
STALE_SECONDS = 120 if HOOK_URL else 10  # Read tasks still open after this
TASK_DEADLINE = 900  # Stop waiting for a task that runs longer than this

db = sqlite3.connect("scrape.db", isolation_level=None)
db.executescript(
    """
CREATE TABLE IF NOT EXISTS jobs (url TEXT PRIMARY KEY, domain TEXT,
  state TEXT DEFAULT 'queued', step INT DEFAULT 0, retries INT DEFAULT 0,
  credits REAL DEFAULT 0, task_id TEXT, outcome TEXT,
  next_at REAL DEFAULT 0, updated REAL DEFAULT 0, started REAL DEFAULT 0);
CREATE TABLE IF NOT EXISTS domains (domain TEXT PRIMARY KEY,
  paused_until REAL DEFAULT 0, blocks INT DEFAULT 0,
  best_step INT DEFAULT 0, best_until REAL DEFAULT 0);
CREATE TABLE IF NOT EXISTS events (task_id TEXT, event TEXT,
  PRIMARY KEY (task_id, event));
"""
)


class RetryBudget:
    """Per-process token bucket: each first attempt earns 0.1 retry."""

    def __init__(self, ratio=0.1, cap=10.0):
        self.ratio, self.cap, self.tokens = ratio, cap, cap

    def earn(self):
        self.tokens = min(self.cap, self.tokens + self.ratio)

    def spend(self):
        if self.tokens < 1:
            return False
        self.tokens -= 1
        return True


budget = RetryBudget()


def log(outcome, mode, url):
    print(f"{outcome:<13} {mode:<8} {url}", flush=True)


def add_urls(urls):
    for url in urls:
        domain = urlsplit(url).hostname
        if not domain:
            log("skipped", "-", url)
            continue
        db.execute(
            "INSERT OR IGNORE INTO jobs (url, domain) VALUES (?, ?)", (url, domain)
        )
        db.execute("INSERT OR IGNORE INTO domains (domain) VALUES (?)", (domain,))


def retry_after_seconds(value, default=300, cap=3600):
    if not value:
        return default
    if value.strip().isdigit():
        return min(int(value), cap)
    try:
        when = email.utils.parsedate_to_datetime(value)  # The HTTP-date form
    except (TypeError, ValueError):
        return default
    return min(max(when.timestamp() - time.time(), 0), cap)


def record_result(url, job):
    """Turn a finished task into the job's next state."""
    domain, step, retries = db.execute(
        "SELECT domain, step, retries FROM jobs WHERE url=?", (url,)
    ).fetchone()
    outcome, now = classify(job, has_text), time.time()
    log(outcome, STEPS[step]["mode"], url)
    new = {"state": "done", "outcome": outcome, "task_id": None}
    if outcome == "ok":
        save(url, job["content"])
    if outcome in ("ok", "gone"):
        db.execute(
            "UPDATE domains SET blocks=0, best_step=?, best_until=?" " WHERE domain=?",
            (step, now + LEARN_SECONDS, domain),
        )
    elif outcome in ("blocked", "invalid"):
        if step + 1 < len(STEPS):
            new.update(state="queued", step=step + 1, retries=0)
        else:  # Only a block on the last step counts toward a pause
            db.execute("UPDATE domains SET blocks=blocks+1 WHERE domain=?", (domain,))
            db.execute(
                "UPDATE domains SET paused_until=?, blocks=0"
                " WHERE domain=? AND blocks>=?",
                (now + PAUSE_SECONDS, domain, PAUSE_AFTER_BLOCKS),
            )
    elif outcome == "throttled":
        wait = retry_after_seconds(target_headers(job).get("retry-after"))
        db.execute(
            "UPDATE domains SET paused_until=? WHERE domain=?", (now + wait, domain)
        )
        if retries < MAX_RETRIES:  # After that, keep the outcome
            new.update(state="queued", retries=retries + 1, next_at=now + wait)
    elif (
        outcome in ("api_error", "target_error")
        and retries < MAX_RETRIES
        and budget.spend()
    ):  # Over the retry budget: keep the outcome
        new.update(state="queued", retries=retries + 1, next_at=now + backoff(retries))
    sets = ", ".join(f"{column}=?" for column in new)  # Names from code only
    db.execute(
        f"UPDATE jobs SET {sets}, credits=credits+?, updated=?" " WHERE url=?",
        (*new.values(), float(job.get("credits_used") or 0), now, url),
    )


async def read_task(client, task_id):
    """Return ('done', job), ('running', None) or ('lost', None)."""
    try:
        r = await client.get(f"{BASE}/tasks/{task_id}", headers=HEADERS)
    except httpx.TransportError:
        return "running", None  # Try again on the next poll
    if r.headers.get("x-task-finished-at"):
        return "done", r.json()
    if r.status_code == 404:
        return "lost", None
    if r.status_code in (401, 402):
        r.raise_for_status()  # Same stop as a 401 or 402 on submit
    return "running", None  # 202 while processing, or a temporary error


async def finish(client, task_id):
    row = db.execute(
        "SELECT url, started FROM jobs WHERE task_id=? AND" " state='running'",
        (task_id,),
    ).fetchone()
    if not row:
        return  # Not running: processed already, or task_id not saved yet
    url, started = row
    status, job = await read_task(client, task_id)
    if not db.execute(
        "SELECT 1 FROM jobs WHERE task_id=? AND" " state='running'", (task_id,)
    ).fetchone():
        return  # Another reader processed it during the GET
    if status == "done":
        record_result(url, job)
    elif status == "lost":  # No record for this task: run the URL again
        db.execute(
            "UPDATE jobs SET state='queued', task_id=NULL" " WHERE url=?", (url,)
        )
    elif time.time() - started > TASK_DEADLINE:
        db.execute(
            "UPDATE jobs SET state='done', outcome='no_result',"
            " task_id=NULL WHERE url=?",
            (url,),
        )
    else:
        db.execute("UPDATE jobs SET updated=? WHERE url=?", (time.time(), url))


async def webhook(request):
    raw = await request.read()  # Verify the exact bytes that were signed
    sent = request.headers.get("X-Webhook-Signature") or request.headers.get(
        "X-Evomi-Signature", ""
    )
    good = "sha256=" + hmac.new(SECRET.encode(), raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sent, good):
        return web.Response(status=401)
    event = json.loads(raw)
    task_id, kind = event["data"]["task_id"], event["event"]
    first = db.execute(
        "INSERT OR IGNORE INTO events VALUES (?, ?)", (task_id, kind)
    ).rowcount
    if first and kind in ("completed", "failed"):
        # Reply now, and read the task's content in the background
        task = asyncio.create_task(finish(request.app["client"], task_id))
        request.app["pending"].add(task)
        task.add_done_callback(request.app["pending"].discard)
    return web.Response(text="ok")


async def submit(client, url):
    domain, step, retries, credits = db.execute(
        "SELECT domain, step, retries, credits FROM jobs WHERE url=?", (url,)
    ).fetchone()
    best, until = db.execute(
        "SELECT best_step, best_until FROM domains" " WHERE domain=?", (domain,)
    ).fetchone()
    first = step == 0 and retries == 0 and credits == 0
    if first and until > time.time():
        step = best  # Start where this domain last succeeded
    mode = STEPS[step]["mode"]
    if credits + COST[mode] > CREDIT_BUDGET:
        db.execute(
            "UPDATE jobs SET state='done', outcome='over_budget'" " WHERE url=?", (url,)
        )
        log("over_budget", mode, url)
        return
    db.execute(
        "UPDATE jobs SET state='submitting', step=?, updated=?" " WHERE url=?",
        (step, time.time(), url),
    )
    body = {
        "url": url,
        **STEPS[step],
        "async": True,
        "delivery": "json",
        "include_content": True,
        "capture_headers": True,
    }
    if HOOK_URL:
        body["webhook"] = {
            "url": HOOK_URL,
            "type": "custom",
            "secret": SECRET,
            "events": ["completed", "failed"],
        }
    try:
        r = await client.post(f"{BASE}/realtime", headers=HEADERS, json=body)
    except (httpx.ConnectError, httpx.ConnectTimeout):  # It was never sent
        db.execute(
            "UPDATE jobs SET state='queued', next_at=? WHERE url=?",
            (time.time() + 5, url),
        )
        return
    except httpx.TransportError:
        return  # A task may exist, so the poller requeues it later
    if r.status_code in (429, 503):  # Rejected, so no task exists
        db.execute(
            "UPDATE jobs SET state='queued', next_at=? WHERE url=?",
            (time.time() + 5, url),
        )
        return
    if r.status_code in (400, 422):  # Includes a domain that doesn't exist
        db.execute(
            "UPDATE jobs SET state='done', outcome='rejected'" " WHERE url=?", (url,)
        )
        log("rejected", mode, url)
        return
    if r.status_code >= 500:
        return  # Unknown outcome: the poller requeues it later
    if r.status_code in (401, 402):  # No task exists: undo 'submitting'
        db.execute("UPDATE jobs SET state='queued' WHERE url=?", (url,))
    r.raise_for_status()  # A 401 or 402 stops the service on purpose
    now = time.time()
    db.execute(
        "UPDATE jobs SET state='running', task_id=?, updated=?,"
        " started=? WHERE url=?",
        (r.json()["task_id"], now, now, url),
    )
    if first:
        budget.earn()  # Each URL's first accepted attempt adds 0.1 retry


async def dispatcher(client):
    """Submit queued jobs inside the slot limit. Return when none are open."""
    while True:
        now = time.time()
        open_jobs = db.execute(
            "SELECT count(*) FROM jobs WHERE state IN"
            " ('queued', 'running', 'submitting')"
        ).fetchone()[0]
        if not open_jobs:
            for outcome, count in db.execute(
                "SELECT outcome, count(*) FROM" " jobs GROUP BY outcome"
            ):
                print(f"done: {count} x {outcome}", flush=True)
            return
        busy = db.execute(
            "SELECT count(*) FROM jobs WHERE state IN" " ('running', 'submitting')"
        ).fetchone()[0]
        rows = db.execute(
            """
            SELECT j.url FROM jobs j JOIN domains d USING (domain)
            WHERE j.state='queued' AND j.next_at<=? AND d.paused_until<=?
            ORDER BY j.next_at LIMIT ?""",
            (now, now, max(CONCURRENCY - busy, 0)),
        ).fetchall()
        for (url,) in rows:
            await submit(client, url)
        await asyncio.sleep(1)


async def poller(client):
    """Read tasks that are still open, and recover after a crash."""
    while True:
        stale = time.time() - STALE_SECONDS
        # A crash between the POST and saving task_id leaves 'submitting'.
        # Requeue those jobs as a retry, and print each URL.
        db.execute(
            "UPDATE jobs SET state='done', outcome='no_result'"
            " WHERE state='submitting' AND updated<? AND retries>=?",
            (stale, MAX_RETRIES),
        )
        for (url,) in db.execute(
            "SELECT url FROM jobs WHERE" " state='submitting' AND updated<?", (stale,)
        ).fetchall():
            log("resubmitted", "-", url)
        db.execute(
            "UPDATE jobs SET state='queued', retries=retries+1"
            " WHERE state='submitting' AND updated<?",
            (stale,),
        )
        rows = db.execute(
            "SELECT task_id FROM jobs WHERE state='running'" " AND updated<?", (stale,)
        ).fetchall()
        for (task_id,) in rows:
            await finish(client, task_id)
        await asyncio.sleep(30 if HOOK_URL else 5)


async def main(urls):
    add_urls(urls)
    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
        if HOOK_URL:  # Serve the receiver behind your HTTPS reverse proxy
            app = web.Application()
            app["client"], app["pending"] = client, set()
            app.router.add_post("/webhooks/evomi", webhook)
            runner = web.AppRunner(app)
            await runner.setup()
            await web.TCPSite(runner, "127.0.0.1", PORT).start()
        tasks = {
            asyncio.create_task(poller(client)),
            asyncio.create_task(dispatcher(client)),
        }
        done, rest = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in rest:
            task.cancel()
        for task in done:
            task.result()  # Re-raise a 401 or 402 from either loop


if __name__ == "__main__":
    try:
        with open("urls.txt") as file:
            urls = file.read().split()
    except FileNotFoundError:
        raise SystemExit("Put the URLs to scrape in urls.txt, one per line") from None
    try:
        asyncio.run(main(urls))
    except httpx.HTTPStatusError as exc:  # A 401 or 402 stops the service
        raise SystemExit(
            f"Service stopped: {exc.response.status_code} " f"{exc.response.text[:200]}"
        ) from None

uv run scrape_service.py

blocked       request  https://www.walmart.com/search?q=coffee+maker
target_error  request  https://httpbin.io/status/503
gone          request  https://www.python.org/this-page-does-not-exist
blocked       request  https://www.walmart.com/search?q=espresso+machine
throttled     request  https://httpbin.io/status/429
target_error  request  https://httpbin.io/status/503
ok            browser  https://www.walmart.com/search?q=coffee+maker
ok            browser  https://www.walmart.com/search?q=espresso+machine
target_error  request  https://httpbin.io/status/503
throttled     request  https://httpbin.io/status/429
throttled     request  https://httpbin.io/status/429
done: 1 x gone
done: 2 x ok
done: 1 x target_error
done: 1 x throttled

ok            browser  https://www.walmart.com/search?q=air+fryer
ok            request  https://webhook.site/583855e9-c9b5-4aa6-b8d4-c89e6b74aa93
done: 1 x gone
done: 4 x ok
done: 1 x target_error
done: 1 x throttled