Tutorial 17 min read

Build a Scraping Pipeline You Can Safely Restart Mid-Run

A tutorial on checkpointing batch scraping runs with run manifests and idempotent writes, so a crashed pipeline resumes exactly where it stopped.

FineData Engineering · Editorial Policy

| October 1, 2026

On this page

Why Long Scraping Runs Fail — and Why Restartability Matters

Long scraping runs fail in the middle. A 40,000-page crawl hits a network blip at hour three, the process dies, and now you have a half-written CSV, no idea which pages were actually fetched, and a choice between re-scraping everything (expensive) or guessing at the gap (corrupt data). Neither option is acceptable when you’re paying per successful page.

The fix is boring engineering: a run manifest on disk, atomic checkpoint writes, and idempotent output keyed on stable identifiers. None of it is clever. All of it is the difference between “restart the run and lose 20 minutes” and “restart the run and lose everything since the last flush.”

This tutorial builds a checkpointed pipeline against a batch scraping API, using store.example.com product listings as the target. The pattern works regardless of what fetches your pages — a managed API, your own Playwright fleet, or plain requests. I’ll use the batch endpoints here because they map cleanly onto the chunk model, but the checkpointing logic is fetch-agnostic.

Anatomy of a Restartable Run: What State Must Survive a Crash

A pipeline that can resume needs exactly three pieces of durable state. Not a database with fourteen tables — three pieces:

The run manifest — one JSON file describing the run: what it’s crawling, how far it’s gotten, and the status of every work unit.
Per-item results — the actual scraped data, written keyed by identity (not appended), so re-writing a row is harmless.
The frozen config block — the exact settings the run started with, recorded at start and enforced on resume. It lives inside the manifest file, but it’s conceptually its own piece of state: it’s what makes a resumed run produce the same dataset an uninterrupted run would have.

Everything else can be reconstructed. The on-disk layout looks like this:

runs/
└── run-8f3a2c/                  # run_id: timestamp + short random suffix
    ├── manifest.json            # THE source of truth for progress
    ├── discovery.json           # cached page counts from the discovery pass
    ├── results/
    │   └── products.db          # SQLite, keyed writes (chunk_id, product_id)
    └── checkpoints/
        └── chunk-000042.json    # optional: raw response snapshots per chunk

The manifest schema:

{
  "run_id": "run-8f3a2c",
  "target": "store.example.com",
  "started_at": "2025-01-14T09:02:11Z",
  "chunks": {
    "category=electronics&page=3": "done",
    "category=electronics&page=4": "in_progress",
    "category=electronics&page=5": "pending"
  },
  "chunk_rows": {
    "category=electronics&page=3": 44
  },
  "item_status": {
    "category=electronics&page=4": {
      "https://store.example.com/p/sku-88123": {"status": "done", "attempts": 1},
      "https://store.example.com/p/sku-88124": {"status": "failed", "attempts": 3},
      "https://store.example.com/p/sku-88125": {"status": "pending", "attempts": 0}
    }
  },
  "counts": {"done": 41, "failed": 3, "pending": 118},
  "config": {
    "proxy_country": "US",
    "use_residential": true,
    "formats": ["markdown"]
  }
}

Note the chunk_rows map: for every completed chunk, the manifest records how many rows it wrote. That’s what lets the verification pass at the end of the run cross-check the manifest against the output table — more on that later.

Two design decisions worth defending:

The manifest is one file, not a database. At the scale of tens of thousands of work units, a single JSON file you rewrite atomically is simpler and faster than any database, and you can cat it to debug. If your run has millions of units, shard the manifest per category. Don’t reach for Postgres until a JSON file actually hurts.

The config block is frozen at run start. If you resume a run with different scrape settings than it started with, your dataset is silently heterogeneous. Record the config, refuse to resume if the config changed. This has saved me from merging data fetched through different proxy tiers more than once.

Modeling the Work Queue: Chunking store.example.com Listings into Resumable Units

You cannot checkpoint what you cannot address. Before anything else, enumerate the crawl space into discrete, named units. For a product catalog, the natural unit is a category page: one URL that lists ~48 products.

import json
import os
import re
import requests

API = "https://api.finedata.ai"
HEADERS = {"Authorization": "Bearer fd_your_api_key"}

CATEGORIES = [
    "electronics", "home-kitchen", "sports",
    "books", "toys", "grocery",
]

def fetch_listing_page(listing_url: str) -> dict:
    """Fetch one listing page via the scraping API; return {'total_pages': N}."""
    resp = requests.post(
        f"{API}/api/v1/scrape",
        headers=HEADERS,
        json={"url": listing_url, "formats": ["markdown"]},
        timeout=180,
    )
    resp.raise_for_status()
    markdown = resp.json()["results"][0]["content"]["markdown"]
    m = re.search(r"page 1 of (\d+)", markdown)   # site-specific pagination marker
    return {"total_pages": int(m.group(1)) if m else 1}

def enumerate_chunks(run_dir: str, base_url: str = "https://store.example.com"):
    """Return every (chunk_id, url) pair for the full crawl space.

    Discovery itself is checkpointed: page counts are cached in
    discovery.json, so a resumed run never re-pays for them.
    """
    cache_path = os.path.join(run_dir, "discovery.json")
    if os.path.exists(cache_path):
        with open(cache_path) as f:
            page_counts = json.load(f)
    else:
        page_counts = {}
        for cat in CATEGORIES:
            first = fetch_listing_page(f"{base_url}/c/{cat}?page=1")
            page_counts[cat] = first["total_pages"]
        tmp_path = cache_path + ".tmp"
        with open(tmp_path, "w") as f:
            json.dump(page_counts, f)
        os.replace(tmp_path, cache_path)   # temp-and-rename, explained next section

    chunks = []
    for cat in CATEGORIES:
        for page in range(1, page_counts[cat] + 1):
            chunk_id = f"category={cat}&page={page}"
            chunks.append((chunk_id, f"{base_url}/c/{cat}?page={page}"))
    return chunks

The chunk ID is derived from the URL’s meaning, not its position in a list. That matters: if discovery returns categories in a different order on resume (site added a category, pagination shifted), position-based IDs silently misalign, while semantic IDs stay stable.

Note that the discovery pass is checkpointed too — discovery.json survives crashes just like the manifest. This isn’t optional for any strategy: whatever your chunking granularity, an un-checkpointed discovery phase means every restart re-pays for the initial listing fetches before it can do anything useful.

Which chunking strategy should you pick? It depends on your restart granularity needs:

Strategy	Restart granularity	Manifest size	Duplicate-fetch risk
Per-URL (every product page is a unit)	Single page	Huge (100k+ entries)	Lowest — discovery checkpointing is mandatory
Per-page-range (e.g. 50 listing pages per unit)	50 pages lost per crash	Tiny	Medium — a half-finished range re-fetches up to 49 pages
Per-category-page (one listing page per unit)	One listing page	Moderate (~1-2k entries)	Low — only the in-flight page re-fetches

I recommend per-category-page for catalogs of this shape. Per-URL gives finer granularity but the manifest balloons, and you end up maintaining item-level status for the entire crawl space instead of just the in-flight chunk. Per-page-range is fine for enormous crawls where you accept losing some work, but it wastes the most paid fetches on restart — the exact thing we’re trying to avoid.

One honest trade-off: chunking by listing page means a crash mid-chunk loses the whole in-flight listing page unless you track per-item status inside it. We handle that later with the item_status map.

Writing the Checkpoint: Persisting Progress After Every Completed Chunk

Here is where most homegrown pipelines fail. They write the manifest with a plain json.dump to the final path. If the process dies between open() and the write completing — or the OS hasn’t flushed the page cache — you get a truncated or empty manifest. Now the source of truth is gone, and the resume path has nothing to load.

The fix is the temp-and-rename pattern, and it is non-negotiable:

import json
import os
import tempfile

def write_manifest_atomic(run_dir: str, manifest: dict):
    """Write manifest.json so readers always see a complete file."""
    path = os.path.join(run_dir, "manifest.json")
    fd, tmp_path = tempfile.mkstemp(
        dir=run_dir, prefix=".manifest-", suffix=".tmp"
    )
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(manifest, f, indent=2)
            f.flush()
            os.fsync(f.fileno())          # bytes hit the disk, not just page cache
        os.replace(tmp_path, path)        # atomic on POSIX, same filesystem
        dir_fd = os.open(run_dir, os.O_RDONLY)
        try:
            os.fsync(dir_fd)              # the rename itself is durable
        finally:
            os.close(dir_fd)
    except BaseException:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)
        raise

The directory fsync matters more than it looks: without it, the rename can be lost on a power cut even though the file data survived. Process death doesn’t need it — the rename is already visible to any new process — but power loss does, and the two lines that buy you that guarantee cost nothing.

Why this works: os.replace() is atomic at the filesystem level. Any reader — including a reader that starts while the write is in flight — sees either the old complete manifest or the new complete manifest. Never a partial one.

Walk through the crash scenarios:

Crash before os.replace: manifest.json still holds the previous checkpoint. You lose one chunk of progress. The temp file is orphaned garbage; sweep .manifest-*.tmp files on startup.
Crash during os.replace: impossible to observe a partial state — the syscall either completed or didn’t.
Crash after os.replace but before directory fsync: only matters for power loss, not process death. Process death leaves the rename visible.

The cost is one small JSON rewrite plus two fsyncs per chunk. For a chunk that takes several seconds of network time, this overhead is noise. If you’re checkpointing after every single item, fsync cost starts to matter — batch your checkpoints instead.

Idempotent Writes: Designing Output So Re-Runs Never Duplicate Rows

Checkpoints protect progress. Idempotent writes protect the data itself. The rule: every output row must be addressable by a stable key, and re-writing that key must be a no-op or an overwrite — never an append.

SQLite makes this trivial:

CREATE TABLE IF NOT EXISTS products (
    chunk_id    TEXT NOT NULL,
    product_id  TEXT NOT NULL,
    url         TEXT NOT NULL,
    title       TEXT,
    price       REAL,
    currency    TEXT,
    fetched_at  TEXT NOT NULL,
    PRIMARY KEY (chunk_id, product_id)
);

-- Safe to run any number of times:
INSERT OR REPLACE INTO products
    (chunk_id, product_id, url, title, price, currency, fetched_at)
VALUES
    ('category=electronics&page=3', 'sku-88123',
     'https://store.example.com/p/sku-88123',
     'Wireless Headphones X2', 89.99, 'USD', '2025-01-14T09:14:32Z');

Compare the two writer implementations:

# NAIVE: appends. A restart re-fetches completed chunks and
# duplicates every row. Your price history is now garbage.
def write_naive(row, csv_path):
    with open(csv_path, "a") as f:
        f.write(",".join(str(row[k]) for k in COLUMNS) + "\n")

# IDEMPOTENT: keyed. Re-running a chunk overwrites its own rows.
def write_idempotent(conn, chunk_id: str, products: list):
    conn.executemany(
        """INSERT OR REPLACE INTO products
           (chunk_id, product_id, url, title, price, currency, fetched_at)
           VALUES (:chunk_id, :product_id, :url, :title,
                   :price, :currency, :fetched_at)""",
        [{**p, "chunk_id": chunk_id} for p in products],
    )
    conn.commit()

The naive writer doesn’t just duplicate — it corrupts. If your downstream analysis computes average price per product, duplicates skew it silently. There’s no error, no exception, just wrong numbers. That’s the worst failure mode a pipeline can have, and an append-only writer invites it on every restart.

A note on INSERT OR REPLACE versus INSERT ... ON CONFLICT DO UPDATE: functionally similar here. I use OR REPLACE because it deletes-then-inserts, which handles schema drift where new columns appear — old rows get fully rewritten rather than carrying stale NULLs. If you need to preserve columns not present in the re-run (like a manually added annotation), use ON CONFLICT DO UPDATE with explicit column lists instead.

Resuming the Run: Loading the Manifest and Skipping Completed Work

With the manifest and keyed writes in place, resume becomes almost anticlimactic:

import json, logging, os

log = logging.getLogger("pipeline")

def load_or_init_run(run_dir: str, all_chunks: list[str]) -> dict:
    manifest_path = os.path.join(run_dir, "manifest.json")
    if os.path.exists(manifest_path):
        with open(manifest_path) as f:
            manifest = json.load(f)
        pending = [c for c in all_chunks
                   if manifest["chunks"].get(c) != "done"]
        log.info(
            "resuming at chunk %s of %s (%d done, %d pending)",
            pending[0] if pending else "-",
            len(all_chunks),
            len(all_chunks) - len(pending),
            len(pending),
        )
        return manifest
    return init_manifest(run_dir, all_chunks)

The full pipeline loop ties it together. Each chunk is one batch submission to the scraping API, followed by keyed writes and a checkpoint:

import sqlite3
import time

# API, HEADERS, and requests come from the chunking section above.

def process_chunk(chunk_id: str, listing_url: str, manifest: dict,
                  run_dir: str, conn) -> bool:
    resp = requests.post(
        f"{API}/api/v1/async/batch",
        headers=HEADERS,
        json={
            "requests": [{
                "url": listing_url,
                "formats": ["markdown"],
                "use_residential": True,
                "proxy_country": "US",
                "max_retries": 3,
            }],
        },
        timeout=30,
    )
    resp.raise_for_status()
    batch_id = resp.json()["batch_id"]

    # Poll until the batch finishes
    while True:
        status = requests.get(
            f"{API}/api/v1/async/batch/{batch_id}",
            headers=HEADERS,
            params={"include_results": True},
            timeout=30,
        ).json()
        if status["status"] in ("completed", "failed", "cancelled"):
            break
        time.sleep(5)

    products = parse_products(status["results"][0])
    write_idempotent(conn, chunk_id, products)
    manifest["chunks"][chunk_id] = "done"
    manifest["chunk_rows"][chunk_id] = len(products)
    write_manifest_atomic(run_dir, manifest)
    return True

def run(run_dir: str, all_chunks: list):
    manifest = load_or_init_run(run_dir, all_chunks)
    conn = sqlite3.connect(os.path.join(run_dir, "results", "products.db"))
    setup_schema(conn)

    for chunk_id, url in all_chunks:
        if manifest["chunks"].get(chunk_id) == "done":
            continue                      # already fetched — skip, don't pay again
        process_chunk(chunk_id, url, manifest, run_dir, conn)

Now the crash demo. Kill it mid-run with Ctrl-C, then start it again:

$ python pipeline.py --run runs/run-8f3a2c
INFO: starting run, 1247 chunks total
INFO: chunk category=electronics&page=1 done (44 products)
INFO: chunk category=electronics&page=2 done (48 products)
...
INFO: chunk category=electronics&page=41 done (46 products)
^C
Traceback (most recent call last): KeyboardInterrupt

$ python pipeline.py --run runs/run-8f3a2c --resume
INFO: resuming at chunk category=electronics&page=42 of 1247 (41 done, 1206 pending)
INFO: chunk category=electronics&page=42 done (47 products)
INFO: chunk category=electronics&page=43 done (48 products)
...

The log line is the proof: pages 1 through 41 are never re-submitted. The only fetch you pay for twice is the chunk that was in flight when the interrupt landed — and thanks to keyed writes, even that one can’t duplicate rows. If you want to cut that last bit of waste too, the next section handles it.

If you’re weighing sync versus async submission for the chunks themselves, I wrote a separate comparison of sync vs async scraping trade-offs — the short version is that batches win once a run exceeds a few hundred pages, which is exactly when restartability starts to matter.

Handling Partial Failures Inside a Chunk: Retries Without Losing Completed Items

The messy case: a chunk contains 48 product URLs, you’ve fetched 31 of them, and the process dies. Chunk-level status says in_progress — resume would re-fetch all 48. For expensive runs, track items individually.

The manifest gains an item_status map per in-flight chunk. Each entry is a small dict holding both the status and the attempt count, because the attempt count is exactly what a resumed process needs to avoid quietly doubling its retry budget:

"item_status": {
  "category=electronics&page=4": {
    "https://store.example.com/p/sku-88120": {"status": "done", "attempts": 1},
    "https://store.example.com/p/sku-88121": {"status": "done", "attempts": 1},
    "https://store.example.com/p/sku-88122": {"status": "failed", "attempts": 3},
    "https://store.example.com/p/sku-88123": {"status": "pending", "attempts": 1},
    "https://store.example.com/p/sku-88124": {"status": "pending", "attempts": 0}
  }
}

Then wrap each item fetch in a bounded retry. The critical behavior: after max_retries attempts, mark the item failed and move on. A permanently dead URL must never block the run:

import time

def fetch_item_with_retry(url: str, chunk_id: str, item_status: dict,
                          max_retries: int = 3, base_delay: float = 2.0):
    attempts = item_status.get(url, {}).get("attempts", 0)

    for attempt in range(attempts, max_retries):
        try:
            result = requests.post(
                f"{API}/api/v1/scrape",
                headers=HEADERS,
                json={
                    "url": url,
                    "formats": ["markdown"],
                    "use_residential": True,
                    "proxy_country": "US",
                    "max_retries": 2,
                },
                timeout=180,
            )
            result.raise_for_status()
            item_status[url] = {"status": "done", "attempts": attempt + 1}
            return result.json()
        except requests.HTTPError as exc:
            code = exc.response.status_code
            if 400 <= code < 500 and code != 429:
                # Non-retryable: the destination said no. Mark and move on.
                item_status[url] = {"status": "failed", "attempts": attempt + 1}
                log.error("non-retryable HTTP %d for %s — marking failed",
                          code, url)
                return None
            delay = base_delay * (2 ** attempt)     # 2s, 4s, 8s
            log.warning("attempt %d failed for %s: HTTP %d — retrying in %ss",
                        attempt + 1, url, code, delay)
            item_status[url] = {"status": "pending", "attempts": attempt + 1}
            time.sleep(delay)
        except (requests.RequestException, KeyError) as exc:
            delay = base_delay * (2 ** attempt)     # 2s, 4s, 8s
            log.warning("attempt %d failed for %s: %s — retrying in %ss",
                        attempt + 1, url, exc, delay)
            item_status[url] = {"status": "pending", "attempts": attempt + 1}
            time.sleep(delay)

    item_status[url] = {"status": "failed", "attempts": max_retries}
    log.error("giving up on %s after %d attempts", url, max_retries)
    return None

Note the exception ordering: HTTPError is a subclass of RequestException, so it’s caught first and inspected — 4xx responses (except 429, which deserves one backoff-and-retry) fail immediately, while 5xx responses fall through to the backoff path alongside timeouts and connection errors.

Two opinions baked into that code:

Don’t retry non-retryable failures. A 404 or a 403 from the destination is not going to fix itself in 8 seconds. Blind exponential backoff on a 404 wastes minutes per dead URL, and a crawl with even a small fraction of dead links turns that waste into hours.

Record the attempt count in the manifest, not in memory. If the crash happens during the backoff sleep, the resumed process needs to know how many attempts already burned. Otherwise your “3 retries” quietly becomes 6 or 9 across restarts.

The failed-items list is a deliverable, not just an internal state. At run end, dump every failed URL to a file. Feed it to a re-drive pass later — the same pattern covered in detail in handling partial batch results. A run with a 2% failure rate and a recorded failure list beats a run that silently pretends it covered everything.

Verifying a Clean Resume: Measuring Coverage and Detecting Silent Data Loss

Resume logic that you haven’t tested is a hypothesis. Build a verification pass that runs automatically at the end of every run — resumed or not — and answers one question: does the output table contain exactly what the manifest says it should?

def verify_run(run_dir: str, manifest: dict, conn) -> bool:
    problems = []

    # 1. Chunk-level coverage: every 'done' chunk must have rows
    for chunk_id, status in manifest["chunks"].items():
        if status != "done":
            continue
        rows = conn.execute(
            "SELECT COUNT(*) FROM products WHERE chunk_id = ?",
            (chunk_id,),
        ).fetchone()[0]
        if rows == 0:
            problems.append(f"chunk {chunk_id} marked done but has 0 rows")

    # 2. Global counts: manifest row counts vs actual table rows
    expected_rows = sum(manifest.get("chunk_rows", {}).values())
    table_rows = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0]
    if table_rows != expected_rows:
        problems.append(
            f"row count mismatch: manifest accounts for {expected_rows} "
            f"rows, table has {table_rows}"
        )

    # 3. Duplicates: impossible by schema, but assert anyway —
    #    someone will eventually "migrate" the table and drop the PK
    dupes = conn.execute(
        """SELECT chunk_id, product_id, COUNT(*) c FROM products
           GROUP BY chunk_id, product_id HAVING c > 1"""
    ).fetchall()
    if dupes:
        problems.append(f"duplicate rows detected: {dupes[:5]}")

    if problems:
        for p in problems:
            log.error("VERIFY FAIL: %s", p)
        return False
    done_chunks = sum(1 for s in manifest["chunks"].values() if s == "done")
    log.info("verify pass: %d done chunks, %d rows, 0 duplicates",
             done_chunks, table_rows)
    return True

Check #2 is the one that catches silent data loss: because process_chunk records chunk_rows[chunk_id] at every checkpoint, the manifest carries an independent accounting of how many rows the run believes it wrote. If the table disagrees — rows lost to a failed commit, a botched migration, a chunk marked done before its write landed — the verify pass says so instead of letting you find out downstream.

That third check looks paranoid. It isn’t. The most common way idempotency dies is a well-meaning schema migration that drops the primary key, after which every restart silently duplicates rows. Assert the invariant; make it impossible to regress unnoticed.

Here’s what a local measurement from a sample run looked like — a deliberately crashed crawl of 300 listing chunks, killed roughly two-thirds through, then resumed:

Metric	Value
Total chunks in run	300
Chunks completed before crash	204
Chunks completed after resume	96
Duplicate rows detected	0
Pages re-fetched on resume	1 (the in-flight chunk)
Wall-clock time lost to crash + restart	~40 seconds

The one re-fetched page is the in-flight chunk whose results hadn’t been written when the interrupt landed. With per-item status tracking inside that chunk, even that goes to zero — at the cost of a manifest write per item, which I don’t think is worth it below ~10k pages per run. Your threshold may differ; measure the fsync overhead against your per-page fetch cost before committing either way.

Wrap-Up

A restartable scraping pipeline needs three things and nothing exotic: a semantic-chunked work queue, an atomically written manifest, and keyed idempotent output. Crash mid-run and you lose one in-flight unit. Crash during a manifest write and you lose nothing, because the temp-and-rename pattern guarantees readers always see a complete file. Re-run any chunk and the output table simply overwrites itself.

The pieces I’d urge you not to skip: the frozen config block (heterogeneous datasets from mismatched resume settings are miserable to diagnose), the automatic verify pass at run end, and the failed-URL dump. Together they turn “probably fine” into a checkable claim.

Next steps worth building on this foundation: swap the polling loop for the callback_url webhook on batch submission so the pipeline is event-driven instead of sleep-polling, and add change detection so re-runs only re-fetch pages that actually changed rather than everything on a schedule. The manifest you already have is the natural place to store content hashes per chunk — the same file that makes restarts safe makes incremental refresh cheap.

#web scraping #data pipeline #checkpointing #idempotency #python #slot:pipeline

Tutorial