Capstone: Build a Production Scraper

Put the Whole Course on One Target#

Every earlier lesson taught one technique in isolation. The capstone forces them into a single system. The assignment: monitor a product catalogue, keep a clean database, fetch only what changed, stay within the site's tolerance, and survive a crash without losing or duplicating data. That sentence is the whole course. Below is the build order, the lesson each step leans on, a working skeleton that ties the pieces together, and a rubric with concrete checks that tell you when you are done.

Choose a target you are allowed to crawl: a site whose robots.txt permits it and whose terms do not forbid it, a sandbox built for scraping practice, or a site you run yourself. A catalogue with a few thousand items and a sitemap is ideal.

Project Layout#

Keep each concern in its own module so every step can be tested alone:

catalog-monitor/
  config.py        # target, rate, user agent, paths
  discover.py      # robots.txt + sitemap -> URL list with lastmod
  fetch.py         # polite session, conditional GET, raw cache
  parse.py         # HTML -> record dict, drift checks
  store.py         # SQLite schema, upsert, run status
  run.py           # entry point: discover -> fetch -> parse -> store
  tests/fixtures/  # saved HTML pages for parser regression tests

Step 1: Reconnaissance#

Read robots.txt, collect its Sitemap: lines, and check for feeds. Open the browser's Network tab on a product page: if the data arrives as JSON from an API, fetch that instead of parsing HTML (see XHR and API reverse engineering). Write down the fields you need and where each one lives. This step often decides the whole design, and the sitemaps and feeds lesson shows why a sitemap with lastmod is the best inventory you can get.

import urllib.robotparser
import xml.etree.ElementTree as ET
import httpx

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

def discover(base, client, agent):
    rp = urllib.robotparser.RobotFileParser(base + "/robots.txt")
    rp.read()
    sitemaps = rp.site_maps() or [base + "/sitemap.xml"]
    urls = {}
    while sitemaps:
        root = ET.fromstring(client.get(sitemaps.pop()).content)
        for loc in root.findall("sm:sitemap/sm:loc", NS):   # sitemap index
            sitemaps.append(loc.text.strip())
        for url in root.findall("sm:url", NS):
            loc = url.findtext("sm:loc", namespaces=NS).strip()
            if rp.can_fetch(agent, loc):
                urls[loc] = url.findtext("sm:lastmod", default=None, namespaces=NS)
    return urls, rp.crawl_delay(agent)

site_maps() returns the Sitemap: lines from robots.txt (or None), and crawl_delay() returns the declared delay for your agent if there is one. Filtering every URL through can_fetch makes the robots.txt rules part of the code instead of a promise.

Step 2: Polite, Resumable Fetching#

One session, one honest User-Agent, a rate limit no faster than the crawl delay, retries with backoff for 429 and 5xx, and a conditional GET using the validators stored from the last run. The rate limiting and politeness, error handling and incremental scraping lessons each cover one of these; here they combine:

import time

class Fetcher:
    def __init__(self, client, delay):
        self.client, self.delay, self.last = client, delay, 0.0

    def get(self, url, etag=None, last_modified=None, tries=4):
        headers = {}
        if etag:
            headers["If-None-Match"] = etag
        if last_modified:
            headers["If-Modified-Since"] = last_modified
        for attempt in range(tries):
            wait = self.delay - (time.monotonic() - self.last)
            if wait > 0:
                time.sleep(wait)
            self.last = time.monotonic()
            try:
                r = self.client.get(url, headers=headers)
            except httpx.TransportError:
                time.sleep(2 ** attempt * 5)
                continue
            if r.status_code in (429, 500, 502, 503, 504):
                retry_after = r.headers.get("Retry-After", "")
                time.sleep(int(retry_after) if retry_after.isdigit() else 2 ** attempt * 5)
                continue
            return r
        raise RuntimeError(f"gave up on {url}")

A 304 response means "unchanged since your copy": no parsing, no write, almost no bandwidth. Save raw HTML for every 200 to a cache directory keyed by URL hash, so parser fixes can be re-run without re-fetching (see caching and resume).

Step 3: Extraction With Drift Checks#

Parse with lxml-backed BeautifulSoup or lxml directly, using CSS for simple targets and XPath where relationships matter. Decode with the right charset before parsing. Normalise every field you will compare later: price to a decimal plus a currency code, dates to ISO 8601, IDs stripped of whitespace. Make required fields fail loudly, as the layout drift detection lesson describes, so a redesign produces an alert rather than a table of Nones.

Before writing any parsing code, save five or six real product pages into tests/fixtures/, including one with a missing price and one out of stock. Write the parser against those files. They become your regression suite.

Step 4: Storage That Is Honest About Keys#

One row per natural key (SKU, or canonical URL if there is no SKU), an upsert that only bumps updated_at when a value actually changed, validators for the next conditional GET, and a run table:

CREATE TABLE IF NOT EXISTS products (
    sku TEXT PRIMARY KEY,
    url TEXT NOT NULL,
    title TEXT NOT NULL,
    price_amount NUMERIC,
    price_currency TEXT,
    in_stock INTEGER,
    etag TEXT,
    last_modified TEXT,
    first_seen TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    checked_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS price_history (
    sku TEXT, price_amount NUMERIC, price_currency TEXT, seen_at TEXT
);
CREATE TABLE IF NOT EXISTS runs (
    run_id TEXT PRIMARY KEY, started TEXT, finished TEXT,
    urls INTEGER, fetched INTEGER, not_modified INTEGER,
    parsed INTEGER, errors INTEGER, status TEXT
);

checked_at changes on every visit, updated_at only when data changed, and price_history gets a row only on a change. That separation lets you answer both "is the crawler alive?" and "how often do prices move?" (see storing data).

Step 5: The Entry Point#

from datetime import datetime, timezone

def as_utc(value):
    # Sitemap lastmod may be a date, or a datetime with "Z" or an offset like +02:00.
    # Text comparison of those forms is wrong, so compare aware datetimes instead.
    try:
        dt = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
    except (AttributeError, ValueError):
        return None                               # missing or unparseable: do not skip
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)      # date-only or naive: assume UTC
    return dt.astimezone(timezone.utc)

def run(run_id):
    stats = dict(urls=0, fetched=0, not_modified=0, parsed=0, errors=0)
    store.start_run(run_id)
    with httpx.Client(headers={"User-Agent": config.AGENT},
                      timeout=30, follow_redirects=True) as client:
        urls, crawl_delay = discover.discover(config.BASE, client, config.AGENT)
        fetcher = fetch.Fetcher(client, max(crawl_delay or 0, config.MIN_DELAY))
        for url, lastmod in urls.items():
            stats["urls"] += 1
            known = store.get_validators(url)
            checked, modified = as_utc(known and known["checked_at"]), as_utc(lastmod)
            if checked and modified and checked >= modified:
                continue  # sitemap says unchanged since our last check
            try:
                r = fetcher.get(url, known and known["etag"],
                                known and known["last_modified"])
                if r.status_code == 304:
                    stats["not_modified"] += 1
                    store.touch(url)
                    continue
                r.raise_for_status()
                stats["fetched"] += 1
                record = parse.product(r.text, url)
                store.upsert(record, r.headers.get("ETag"),
                             r.headers.get("Last-Modified"))
                stats["parsed"] += 1
            except Exception as exc:
                stats["errors"] += 1
                store.log_error(run_id, url, repr(exc))
    store.finish_run(run_id, stats)

The loop is deliberately boring. Every interesting decision lives in a module that can be tested alone, and the loop records what it did whether each URL succeeded or failed. Resuming after a crash is automatic: URLs checked before the crash have a fresh checked_at and are skipped by the lastmod comparison or answered with a cheap 304.

Step 6: Raise the Ceiling Only When Measured#

Start sequential. If a full run is too slow, measure where time goes first: if most of it is the politeness delay, going faster means asking the site for more than it offered, and the answer is to crawl less (better incremental checks), not harder. If the site allows more throughput, add bounded concurrency with a shared rate limiter from the concurrency lesson. Add proxies only if you have a legitimate need and evidence of IP-level throttling at a rate the site's terms allow. Every speed increase should trace back to a measurement.

Step 7: Automate and Watch#

Schedule the entry point with a lock so runs never overlap, and alert on three queries against your own tables (see scheduling and monitoring):

-- staleness: the newest check should be recent
SELECT MAX(checked_at) FROM products;
-- volume collapse: parsed count vs the last 7 runs
SELECT parsed, (SELECT AVG(parsed) FROM (SELECT parsed FROM runs
        ORDER BY started DESC LIMIT 7 OFFSET 1)) AS baseline
FROM runs ORDER BY started DESC LIMIT 1;
-- error spike
SELECT errors * 1.0 / NULLIF(urls, 0) FROM runs ORDER BY started DESC LIMIT 1;

If the target becomes long-lived and large, porting to the Scrapy framework swaps your hand-rolled scheduler, dupe filter, throttling and job persistence for tested equivalents. You will understand what each setting does because you built it once.

The Rubric That Says "Done"#

A scraper that passes all seven checks is production-grade for its scale. One that only passes "it printed data once" is a script that works until the first weekend.

Practice#

Once the rubric passes, add one stretch goal: track prices across two regions with the geo-targeted scraping lesson, deduplicate the same product sold on two sites with entity resolution, or move the job into a container on a schedule following serverless and container scraping.