Backfill, Then Refresh: A Two-Phase Scrape Pipeline

One Script, Two Jobs

Most scraping projects start the same way: someone needs “all the product data from this site,” a script gets written, it works, and then it quietly becomes the wrong tool. The next request is “keep it updated,” and the script gets bolted onto a cron job that re-scrapes everything on every run. Six weeks later you’re burning through your proxy budget re-fetching 500 pages to catch the 4 that changed, and the site’s rate limiter has started noticing you.

The fix is to stop treating “get the historical data” and “keep the data current” as one job. They have different URL scopes, different tolerances for failure, and completely different cost profiles. This post builds a two-phase pipeline in plain Python and SQLite: a checkpointed backfill that runs once, and a hash-based refresh that runs on a schedule through a small persistent job queue. Everything here runs today with no dependencies beyond requests and the standard library.

Why One-Shot Scraping Fails

A backfill is a one-time sweep of the full URL space. A refresh is a recurring, incremental pass over URLs you’ve already seen. When you merge them into one script, the refresh inherits the backfill’s scope — which means every run costs as much as the first run, and any failure mid-loop puts the whole thing in an ambiguous state.

Job type	Run frequency	URL scope	Request volume	Failure cost
Backfill	Once (plus rare restarts)	Full site: sitemap + all catalog pages	Thousands	High — an unresumable failure means starting over
Refresh	Every N hours	Previously seen URLs only	Proportional to corpus size	Low — next run picks up what this one missed
Naive combined	Every N hours	Full site, every run	Thousands, every run	High — worst of both, plus rate-limit exposure

Here’s the naive version, which I’ve seen in production more times than I’d like to admit:

# naive.py -- re-scrapes everything, every run
import requests, sqlite3

conn = sqlite3.connect("listings.db")
urls = [row[0] for row in conn.execute("SELECT url FROM pages")]

for url in urls:  # 500 hits against store.example.com, every single run
    html = requests.get(url, timeout=30).text
    conn.execute(
        "INSERT OR REPLACE INTO pages (url, html) VALUES (?, ?)",
        (url, html),
    )
conn.commit()

Three problems, in order of how soon they’ll hurt: it re-fetches all 500 URLs even when 490 are byte-identical to last run; it rewrites every row, so anything downstream (webhooks, search indexing, diff alerts) fires 500 times instead of 10; and if it dies at URL 300, you have no idea which half of the corpus is stale. The two-phase design addresses each one directly.

Modeling the Target Site

Before either phase runs, you need two things: a way to enumerate URLs, and a way to make sure the same page always maps to the same database key. The second one sounds trivial and isn’t. If your backfill stores https://store.example.com/products/widget?utm_source=newsletter and your refresh sees https://store.example.com/products/widget, you now have two rows for one product and a delta detector that never converges.

Sitemap discovery first, using stdlib only:

import urllib.request
import xml.etree.ElementTree as ET
from urllib.parse import urlparse

NS = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}

def product_urls(sitemap_url):
    """Yield product page URLs from a sitemap index or plain sitemap."""
    with urllib.request.urlopen(sitemap_url, timeout=30) as r:
        tree = ET.fromstring(r.read())
    for loc in tree.findall(".//sm:loc", NS):
        url = loc.text.strip()
        if "/products/" in urlparse(url).path:
            yield url

If the site has no sitemap, or the sitemap lies, you fall back to crawling the catalog pagination — which is exactly what the backfill phase below does anyway. For the trade-offs between the two discovery approaches, sitemap crawling vs link discovery covers it in detail; the short version is that sitemaps are cheaper but paginated crawling catches pages the sitemap forgot.

Now the key scheme:

from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse

TRACKING_PREFIXES = ("utm_", "fbclid", "gclid", "mc_", "ref", "trk")

def normalize_url(url: str) -> str:
    p = urlparse(url)
    host = p.netloc.lower()
    if host.startswith("www."):
        host = host[4:]
    path = p.path.rstrip("/") or "/"
    query = sorted(
        (k, v) for k, v in parse_qsl(p.query)
        if not any(k.lower().startswith(t) for t in TRACKING_PREFIXES)
    )
    return urlunparse(("https", host, path, "", urlencode(query), ""))

Concrete examples of what this does to real URLs:

in:  https://store.example.com/products/widget/
out: https://store.example.com/products/widget

in:  https://www.store.example.com/products/widget?utm_source=newsletter
out: https://store.example.com/products/widget

in:  https://store.example.com/products/widget?color=red&size=m
out: https://store.example.com/products/widget?color=red&size=m

Note the last case: meaningful parameters survive, and sorting makes ?size=m&color=red and ?color=red&size=m collapse to the same key. Decide carefully what counts as “tracking” for your target — if color=red actually serves different content, it’s part of the identity, not noise.

Phase 1: The Backfill Job

The backfill walks the catalog one page at a time and writes a checkpoint row after every page it commits. If the process dies at page 214, the restart resumes at 214. No partial rework, no ambiguity about what’s been collected.

The schema, exactly as the code uses it:

CREATE TABLE IF NOT EXISTS pages (
    url_key      TEXT PRIMARY KEY,
    url         TEXT NOT NULL,
    content_hash TEXT,
    fetched_at  TEXT
);

CREATE TABLE IF NOT EXISTS listings (
    url_key     TEXT NOT NULL,
    title       TEXT,
    price_cents INTEGER,
    PRIMARY KEY (url_key)
);

CREATE TABLE IF NOT EXISTS checkpoints (
    job        TEXT PRIMARY KEY,
    last_page  INTEGER NOT NULL,
    updated_at TEXT NOT NULL
);

The backfill function:

import re, sqlite3, time
import requests

BASE = "https://store.example.com"
HREF_RE = re.compile(r'href="(/products/[^"]+)"')

def parse_product_links(html):
    """Stub parser -- swap the regex for BeautifulSoup or an extract_rules
    payload that matches the real markup."""
    return [BASE + h for h in HREF_RE.findall(html)]

def backfill(conn, session, delay=1.0, max_pages=10_000):
    row = conn.execute(
        "SELECT last_page FROM checkpoints WHERE job = 'backfill'"
    ).fetchone()
    page = row[0] + 1 if row else 1
    if row:
        print(f"resuming backfill from page {page}")

    while page <= max_pages:
        resp = session.get(f"{BASE}/catalog", params={"page": page}, timeout=30)
        if resp.status_code != 200 or "product-card" not in resp.text:
            raise RuntimeError(
                f"catalog page {page} failed or looks like a block page"
            )

        links = parse_product_links(resp.text)
        if not links:
            break  # past the last page

        with conn:  # one transaction: page data + checkpoint commit together
            for url in links:
                key = normalize_url(url)
                conn.execute(
                    "INSERT OR IGNORE INTO pages (url_key, url) VALUES (?, ?)",
                    (key, url),
                )
            conn.execute(
                """INSERT INTO checkpoints (job, last_page, updated_at)
                   VALUES ('backfill', ?, datetime('now'))
                   ON CONFLICT(job) DO UPDATE
                   SET last_page = excluded.last_page,
                       updated_at = excluded.updated_at""",
                (page,),
            )

        page += 1
        time.sleep(delay)

Two details matter more than they look. First, the page data and the checkpoint go into the same transaction — if the commit fails, neither lands, so the checkpoint can never claim a page that wasn’t stored. Second, INSERT OR IGNORE means a re-run after a crash can safely re-see pages it already recorded.

If the catalog blocks plain HTTP clients, swap session.get for a scraping-API call and leave everything else untouched — the checkpoint logic doesn’t care where the HTML came from:

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

def fetch_catalog_page(page):
    r = requests.post(f"{API}/api/v1/scrape", headers=HDRS, json={
        "url": f"{BASE}/catalog?page={page}",
        "formats": ["rawHtml"],
        "use_js_render": True,   # if the catalog is an SPA
        "only_main_content": True,
    }, timeout=120)
    r.raise_for_status()
    return r.json()  # the rawHtml format carries the page HTML

The same swap works for the refresh phase. If you go this route, read the checkpoints-and-restart post first — the failure modes differ slightly when the fetch itself is a billable API call.

Phase 2: The Refresh Job

The refresh walks every URL in pages, fetches it, and rewrites the row only if the content actually changed. One opinionated detail: hash the extracted fields, not the raw HTML. Raw HTML churns constantly — A/B test banners, CSRF tokens, rotating “featured” widgets — and hashing it will flag half your corpus as changed on every run. Hash what you actually store.

import hashlib

TITLE_RE = re.compile(r'<h1 class="product-title">(.*?)</h1>')
PRICE_RE = re.compile(r'data-price="(\d+)"')

def canonical_fields(html):
    """The fields you actually store, sorted for stable ordering."""
    return sorted(TITLE_RE.findall(html)) + sorted(PRICE_RE.findall(html))

def refresh(conn, session, delay=1.0):
    checked = updated = skipped = 0
    rows = conn.execute(
        "SELECT url_key, url, content_hash FROM pages"
    ).fetchall()

    for url_key, url, old_hash in rows:
        resp = session.get(url, timeout=30)
        if resp.status_code != 200:
            skipped += 1  # leave the old row; the next run retries it
            continue

        new_hash = hashlib.sha256(
            "\n".join(canonical_fields(resp.text)).encode()
        ).hexdigest()

        if new_hash != old_hash:
            with conn:
                conn.execute(
                    """UPDATE pages SET content_hash = ?,
                       fetched_at = datetime('now') WHERE url_key = ?""",
                    (new_hash, url_key),
                )
                conn.execute(
                    "DELETE FROM listings WHERE url_key = ?", (url_key,)
                )
                for title, price in zip(
                    TITLE_RE.findall(resp.text), PRICE_RE.findall(resp.text)
                ):
                    conn.execute(
                        "INSERT INTO listings VALUES (?, ?, ?)",
                        (url_key, title, int(price)),
                    )
            updated += 1
        else:
            skipped += 1
        checked += 1
        time.sleep(delay)

    return checked, updated, skipped

A sample run looks like this:

$ python pipeline.py refresh
refresh: checked=40 updated=3 skipped=37 elapsed=41.2s

Be honest with yourself about what this saves. Hashing eliminates rewrites and downstream events — your diff alerts, search reindexing, and webhook consumers only see the 3 changed pages, not all 40. It does not eliminate the fetches: you still have to download every page to compute the hash. If you want to skip fetches too, store the ETag or Last-Modified response header during refresh and send it back as a conditional request header on the next run; sites that honor it return a 304 with no body, and your request cost drops to near zero for unchanged pages. Plenty of sites ignore conditional headers entirely, so measure before you count on it.

Scheduling the Refresh

Cron on your laptop is not a scheduler — it’s a suggestion that only works while the machine is awake. And a Python while True: sleep() loop with the schedule hardcoded inside loses all state on restart. The middle ground is a SQLite-backed job queue: it survives restarts, it retries failures with backoff, and it’s about 60 lines of code.

I’d take this over Celery or RQ for anything under a few hundred jobs a day. You get zero extra infrastructure, the queue state sits in the same database as your data, and debugging means opening a SQLite file. The moment you need distributed workers or per-job priorities, move on — but don’t start there.

The queue table joins the schema from earlier:

CREATE TABLE IF NOT EXISTS jobs (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    kind         TEXT NOT NULL,             -- 'refresh', future kinds
    next_run_at  REAL NOT NULL,
    attempts     INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 5,
    status       TEXT NOT NULL DEFAULT 'pending'  -- pending|running|done|failed
);

The queue module:

import time

def enqueue(conn, kind, run_at=None):
    with conn:
        conn.execute(
            "INSERT INTO jobs (kind, next_run_at) VALUES (?, ?)",
            (kind, run_at or time.time()),
        )

def claim_next(conn):
    now = time.time()
    with conn:
        row = conn.execute(
            """SELECT * FROM jobs
               WHERE status = 'pending' AND next_run_at <= ?
               ORDER BY next_run_at LIMIT 1""",
            (now,),
        ).fetchone()
        if row is None:
            return None
        cur = conn.execute(
            """UPDATE jobs SET status = 'running'
               WHERE id = ? AND status = 'pending'""",
            (row["id"],),
        )
        if cur.rowcount != 1:  # someone else claimed it first
            return None
        return conn.execute(
            "SELECT * FROM jobs WHERE id = ?", (row["id"],)
        ).fetchone()

def mark_done(conn, job_id, reschedule_seconds=None):
    with conn:
        if reschedule_seconds:
            conn.execute(
                """UPDATE jobs SET status = 'pending', attempts = 0,
                   next_run_at = ? WHERE id = ?""",
                (time.time() + reschedule_seconds, job_id),
            )
        else:
            conn.execute("UPDATE jobs SET status = 'done' WHERE id = ?", (job_id,))

def mark_failed(conn, job):
    delay = min(300, 30 * 2 ** job["attempts"])  # 30s, 60s, 120s... capped
    with conn:
        if job["attempts"] + 1 >= job["max_attempts"]:
            conn.execute("UPDATE jobs SET status = 'failed' WHERE id = ?",
                         (job["id"],))
        else:
            conn.execute(
                """UPDATE jobs SET status = 'pending', attempts = attempts + 1,
                   next_run_at = ? WHERE id = ?""",
                (time.time() + delay, job["id"],)
            )

The rowcount != 1 check in claim_next is what makes this safe with multiple worker processes: the claim is a compare-and-swap on the status column, so two workers can never hold the same job.

The worker loop:

def worker(conn, poll_seconds=30, refresh_interval_hours=24):
    session = requests.Session()
    session.headers["User-Agent"] = "Mozilla/5.0 (compatible; pricebot/1.0)"

    while True:
        job = claim_next(conn)
        if job is None:
            time.sleep(poll_seconds)
            continue
        try:
            if job["kind"] == "refresh":
                checked, updated, skipped = refresh(conn, session)
                print(f"refresh: checked={checked} updated={updated} "
                      f"skipped={skipped}")
                mark_done(conn, job["id"],
                          reschedule_seconds=refresh_interval_hours * 3600)
        except Exception as exc:
            print(f"job {job['id']} ({job['kind']}) failed: {exc}")
            mark_failed(conn, job)

After a successful refresh, the job goes back to pending with next_run_at set 24 hours out and attempts reset. After five failures it lands in failed and stays there — which is correct behavior. A refresh that fails five times in a row is telling you the target changed or your account is out of credit, and blind retrying just burns money.

Running Both Phases Today

Everything above ties together behind one argparse entrypoint. Clone, write the config, run.

import argparse, json, sqlite3

def main():
    ap = argparse.ArgumentParser(description="two-phase scrape pipeline")
    ap.add_argument("--config", default="config.json")
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("backfill")
    sub.add_parser("refresh")
    sub.add_parser("worker")
    args = ap.parse_args()

    cfg = json.load(open(args.config))
    conn = sqlite3.connect(cfg["db"])
    conn.row_factory = sqlite3.Row
    init_schema(conn)  # executes the CREATE TABLE statements from earlier

    if args.cmd == "backfill":
        backfill(conn, requests.Session(), delay=cfg["request_delay"])
    elif args.cmd == "refresh":
        c, u, s = refresh(conn, requests.Session(), delay=cfg["request_delay"])
        print(f"refresh: checked={c} updated={u} skipped={s}")
    elif args.cmd == "worker":
        pending = conn.execute(
            "SELECT COUNT(*) FROM jobs WHERE kind='refresh' "
            "AND status IN ('pending','running')"
        ).fetchone()[0]
        if pending == 0:
            enqueue(conn, "refresh")
        worker(conn, refresh_interval_hours=cfg["refresh_interval_hours"])

if __name__ == "__main__":
    main()

The config:

{
  "db": "listings.db",
  "base_url": "https://store.example.com",
  "request_delay": 1.0,
  "refresh_interval_hours": 24
}

The three commands, in the order you’d actually run them:

python pipeline.py backfill   # one-time, resumable
python pipeline.py refresh    # optional manual test of phase 2
python pipeline.py worker     # leave running; owns the schedule from here

The pending == 0 guard before enqueue matters — restarting the worker shouldn’t stack a second refresh job on top of the existing one, or you’ll run the schedule twice as often as intended.

Measuring the Pipeline

You don’t get to claim the two-phase design saves anything until you’ve measured it. Here’s a sample run against a 500-page corpus, with the naive combined scraper included as the baseline:

Metric	Backfill (once)	Refresh run 1	Refresh run 2	Refresh run 3	Naive combined, each run
Pages fetched	500	500	500	500	500
Rows rewritten	500	11	4	9	500
Rows skipped	0	489	496	491	0
Downstream events emitted	500	11	4	9	500
Elapsed	8m 22s	8m 31s	8m 35s	8m 28s	8m 40s

Read that honestly: fetch volume is identical across the refresh runs, because hashing detects change after the fetch. The savings show up in write volume and downstream events — 11 versus 500 is a 45x reduction in what your diff alerts and indexers process. If fetch cost is the thing you need to cut, that’s the conditional-headers play from earlier, or a longer refresh interval, or change-detection heuristics like sitemap lastmod timestamps.

Recovery is the other thing worth verifying. Kill the backfill mid-run and restart it:

$ python pipeline.py backfill
resuming backfill from page 1
...catalog page 213 committed
^CTraceback (most recent call last):
  ...

$ python pipeline.py backfill
resuming backfill from page 214

Prove it from the database rather than trusting the log:

row = conn.execute(
    "SELECT last_page, updated_at FROM checkpoints WHERE job='backfill'"
).fetchone()
print(row["last_page"], row["updated_at"])   # 213  (timestamp of the kill)

The checkpoint says 213, the restart says “from page 214,” and the two agree. That’s the whole contract: the checkpoint records the last committed page, and the resume starts one past it. If you want to go deeper on resumability — handling the case where a page partially committed, or checkpointing at row granularity instead of page granularity — the re-drive partial batch results pattern extends naturally.

Wrapping Up

The two-phase split earns its keep on three fronts: the backfill is resumable so a crash costs you minutes, not the whole run; the refresh touches only what changed, so downstream consumers stop drowning in no-op events; and the job queue keeps the schedule honest across restarts without dragging in a message broker.

Gotchas to watch as you adapt this:

• The URL key scheme is load-bearing. Get normalize_url wrong and your delta detection silently never fires, because every run sees “new” URLs. Test it with the messiest real URLs you can find — tracking params, trailing slashes, mixed-case hosts.
• Hash extracted fields, not raw HTML. Raw-HTML hashing produces constant false positives on any site with rotating boilerplate. You’ll rewrite everything and conclude the design doesn’t work.
• A failed job is a feature, not a bug. Five consecutive refresh failures mean something structural broke. Wire an alert on that status instead of raising max_attempts.
• Respect the target. A 1-second delay across 500 pages is a modest footprint; the same delay across 50,000 pages on one IP is a different conversation. Scale the delay with the corpus, or route through rotation.

Next steps, in order of value: add sitemap lastmod as a pre-filter so the refresh only fetches pages the site itself says changed; move the fetch layer behind a scraping API when plain requests start getting block pages; and add a per-URL cost ledger so you can see exactly what each refresh cycle spends. The structure above holds for all three — none of them touches the checkpoint, hash, or queue logic, which is exactly the point.