Python Web Scraping Project: A Reliable Data Pipeline

TL;DR

A useful Python web scraping project should produce repeatable, validated records rather than stop after printing HTML.
The tutorial below collects the first catalog page from Books to Scrape, normalizes each book into a stable schema, rejects incomplete records, and upserts accepted rows into SQLite.
Timeouts, status checks, bounded retries, stable IDs, and idempotent storage make the difference between a demo script and a maintainable pipeline.
Static HTML is a good fit for Requests and Beautiful Soup; JavaScript-rendered pages require a browser or managed rendering layer.
Only collect authorized or public data, respect applicable terms and privacy obligations, and keep request volume bounded.

Python web scraping project: what you will build

This Python web scraping project builds a small but production-minded pipeline from HTTP response to validated SQLite rows. It uses the purpose-built public practice site Books to Scrape, processes one catalog page, and stores title, price, availability, detail URL, source page, retrieval time, and a stable record ID.

The pipeline has five stages: fetch, parse, normalize, validate, and upsert. Each stage has a clear input and output, so failures can be diagnosed without rerunning unrelated work. The project deliberately stays bounded to one page and does not collect personal or sensitive information.

Nstproxy Crawl is the managed alternative when the same workflow must render JavaScript, crawl a bounded site, or return page artifacts without operating browser workers and extraction infrastructure. For static tutorial HTML, a local Python stack remains the clearest way to learn the mechanics.

Pipeline at a glance

The project transforms one public catalog URL into an idempotent local dataset.

Stage	Input	Operation	Output	Failure boundary
Fetch	URL

from __future__ import annotations

import hashlib
import re
import sqlite3
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

SOURCE_URL = "https://books.toscrape.com/catalogue/page-1.html"
ALLOWED_HOST = "books.toscrape.com"
DB_PATH = "books.db"


@dataclass(frozen=True)
class BookRecord:
    record_id: str
    title: str
    price_gbp: str
    in_stock: bool
    detail_url: str
    source_url: str
    retrieved_at: str


def build_session() -> requests.Session:
    retry = Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=(429, 500, 502, 503, 504),
        allowed_methods=frozenset({"GET"}),
        respect_retry_after_header=True,
    )
    session = requests.Session()
    session.headers["User-Agent"] = "NstproxyTutorialBot/1.0 (bounded demo)"
    session.mount("https://", HTTPAdapter(max_retries=retry))
    return session


def fetch_html(session: requests.Session, url: str) -> str:
    response = session.get(url, timeout=(5, 20))
    response.raise_for_status()
    content_type = response.headers.get("Content-Type", "").lower()
    if "text/html" not in content_type:
        raise ValueError(f"unexpected content type: {content_type}")
    if not response.text.strip():
        raise ValueError("empty HTML response")
    return response.text


def parse_price(text: str) -> Decimal:
    match = re.search(r"\d+(?:\.\d+)?", text.replace(",", ""))
    if not match:
        raise ValueError(f"invalid price: {text!r}")
    try:
        value = Decimal(match.group(0))
    except InvalidOperation as exc:
        raise ValueError(f"invalid price: {text!r}") from exc
    if value <= 0:
        raise ValueError(f"nonpositive price: {value}")
    return value


def parse_books(html: str, source_url: str) -> tuple[list[BookRecord], int]:
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select("article.product_pod")
    if not cards:
        raise ValueError("no product cards found; markup may have changed")

    retrieved_at = datetime.now(timezone.utc).isoformat()
    accepted: list[BookRecord] = []
    rejected = 0

    for card in cards:
        try:
            link = card.select_one("h3 a")
            price_node = card.select_one(".price_color")
            availability_node = card.select_one(".availability")
            if not link or not price_node or not availability_node:
                raise ValueError("missing required node")

            title = (link.get("title") or link.get_text(" ", strip=True)).strip()
            detail_url = urljoin(source_url, link.get("href", ""))
            if not title or urlparse(detail_url).hostname != ALLOWED_HOST:
                raise ValueError("invalid title or detail URL")

            price = parse_price(price_node.get_text(" ", strip=True))
            stable_id = hashlib.sha256(detail_url.encode("utf-8")).hexdigest()
            accepted.append(
                BookRecord(
                    record_id=stable_id,
                    title=title,
                    price_gbp=f"{price:.2f}",
                    in_stock="in stock" in availability_node.get_text(" ", strip=True).lower(),
                    detail_url=detail_url,
                    source_url=source_url,
                    retrieved_at=retrieved_at,
                )
            )
        except (AttributeError, TypeError, ValueError):
            rejected += 1

    return accepted, rejected


def store_books(records: list[BookRecord], db_path: str) -> int:
    with sqlite3.connect(db_path) as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS books (
                record_id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                price_gbp TEXT NOT NULL,
                in_stock INTEGER NOT NULL,
                detail_url TEXT NOT NULL,
                source_url TEXT NOT NULL,
                retrieved_at TEXT NOT NULL
            )
            """
        )
        connection.executemany(
            """
            INSERT INTO books VALUES (
                :record_id, :title, :price_gbp, :in_stock,
                :detail_url, :source_url, :retrieved_at
            )
            ON CONFLICT(record_id) DO UPDATE SET
                title = excluded.title,
                price_gbp = excluded.price_gbp,
                in_stock = excluded.in_stock,
                source_url = excluded.source_url,
                retrieved_at = excluded.retrieved_at
            """,
            [asdict(record) for record in records],
        )
        return connection.execute("SELECT COUNT(*) FROM books").fetchone()[0]


def main() -> None:
    with build_session() as session:
        html = fetch_html(session, SOURCE_URL)
    records, rejected = parse_books(html, SOURCE_URL)
    if not records:
        raise RuntimeError("no valid records accepted")
    total_rows = store_books(records, DB_PATH)
    print(f"accepted={len(records)} rejected={rejected} total_rows={total_rows}")
    print(f"sample={records[0].title!r} price_gbp={records[0].price_gbp}")


if __name__ == "__main__":
    main()

TL;DR

Python web scraping project: what you will build

Pipeline at a glance

Prerequisites

Detailed Tutorial

Method 1: Build a static-HTML scraping pipeline

Step 1: Define the schema and stable ID

Step 2: Fetch with a timeout and bounded retries

Step 3: Parse durable card structure

Step 4: Normalize and validate values

Step 5: Upsert into SQLite

Scale Beyond Static HTML

Step 6: Run the complete project

Method 2: Use managed crawling for rendered or multi-page targets

Testing and acceptance checks

Common failure modes

The selector returns zero cards

Requests time out or receive rate limits

Text contains unexpected characters

Duplicate records appear

The page requires JavaScript

Responsible use

Conclusion

Experience Nstproxy — Start Your Free Trial Today

FAQ