Initial commit: tested Python web scraping examples

Companion code for python-web-scraping.com. Five self-contained example
modules (parsing, regex extraction, resilient HTTP session, validate+store,
async fetching) each with tests. CI runs ruff + pytest on Python 3.10-3.13;
tests are deterministic and offline (local fixtures + httpx MockTransport).

Author: python-web-scraping-com

.github/workflows/ci.yml

@@ -0,0 +1,26 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+      - name: Install
+        run: pip install -e ".[dev]"
+      - name: Lint (ruff)
+        run: ruff check .
+      - name: Test (pytest)
+        run: pytest

.gitignore

@@ -0,0 +1,24 @@
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+.eggs/
+build/
+dist/
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# Test / tooling caches
+.pytest_cache/
+.ruff_cache/
+.coverage
+htmlcov/
+
+# OS / editor
+.DS_Store
+*~
+.idea/
+.vscode/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Python Web Scraping (https://python-web-scraping.com)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,62 @@
+# Python Web Scraping — Examples
+
+[![CI](https://github.com/python-web-scraping-com/examples/actions/workflows/ci.yml/badge.svg)](https://github.com/python-web-scraping-com/examples/actions/workflows/ci.yml)
+[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
+
+Runnable, **tested** Python web scraping examples — the companion code for
+[python-web-scraping.com](https://python-web-scraping.com).
+
+Every example is a small, self-contained function with a test. Each module maps
+to a guide on the site, so you can read the explanation there and run the code here.
+
+## Quick start
+
+```bash
+git clone https://github.com/python-web-scraping-com/examples.git
+cd examples
+python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
+pip install -e ".[dev]"
+```
+
+Run any example directly:
+
+```bash
+python -m pws_examples.parsing
+python -m pws_examples.storage
+python -m pws_examples.async_scrape   # this one makes real HTTP requests
+```
+
+Run the tests and linter:
+
+```bash
+pytest
+ruff check .
+```
+
+## What's inside
+
+| Module | What it shows | Guide |
+| --- | --- | --- |
+| `pws_examples/parsing.py` | Parse product cards and HTML tables with BeautifulSoup | [Parsing HTML with BeautifulSoup](https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/parsing-html-with-beautifulsoup/) |
+| `pws_examples/extraction.py` | Extract emails and phone numbers with regex | [Extracting Data with Regular Expressions](https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/extracting-data-with-regular-expressions/) |
+| `pws_examples/http_client.py` | A `requests` session with retries and backoff | [Understanding HTTP Requests and Responses](https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/understanding-http-requests-and-responses/) |
+| `pws_examples/storage.py` | Validate with Pydantic, store in SQLite, de-duplicate | [Storing and Exporting Scraped Data](https://python-web-scraping.com/scaling-python-web-scrapers/storing-and-exporting-scraped-data/) |
+| `pws_examples/async_scrape.py` | Concurrent fetching with asyncio + HTTPX and a semaphore | [Asynchronous Scraping with asyncio and HTTPX](https://python-web-scraping.com/scaling-python-web-scrapers/asynchronous-scraping-with-asyncio-and-httpx/) |
+
+The tests use local HTML fixtures and `httpx.MockTransport`, so the suite is
+deterministic and runs offline — no live websites are hit in CI.
+
+## Scrape responsibly
+
+These examples are for learning. When scraping real sites, respect `robots.txt`,
+rate-limit your requests, identify your client honestly, and follow each site's
+terms of service.
+
+## Contributing
+
+Contributions are welcome. Please keep each example small and focused, add a test
+for it, and make sure `pytest` and `ruff check .` pass before opening a PR.
+
+## License
+
+[MIT](LICENSE) © Python Web Scraping

pws_examples/__init__.py

@@ -0,0 +1,7 @@
+"""Runnable, tested Python web scraping examples for python-web-scraping.com.
+
+Each module is self-contained and maps to a guide on the site. Run any module
+directly to see it work, e.g. ``python -m pws_examples.parsing``.
+"""
+
+__version__ = "0.1.0"

pws_examples/async_scrape.py

@@ -0,0 +1,44 @@
+"""Asynchronous scraping with asyncio + HTTPX, bounded by a semaphore.
+
+Companion code for:
+- https://python-web-scraping.com/scaling-python-web-scrapers/asynchronous-scraping-with-asyncio-and-httpx/
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+
+
+async def _fetch(client: httpx.AsyncClient, url: str, semaphore: asyncio.Semaphore) -> str | None:
+    async with semaphore:
+        try:
+            response = await client.get(url, timeout=10)
+            response.raise_for_status()
+            return response.text
+        except httpx.HTTPError:
+            return None
+
+
+async def scrape(
+    urls: list[str],
+    concurrency: int = 5,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> list[str | None]:
+    """Fetch many URLs concurrently, capping in-flight requests with a semaphore.
+
+    Failed requests come back as ``None`` rather than cancelling the batch. The
+    ``transport`` parameter lets tests inject ``httpx.MockTransport`` for
+    deterministic, offline runs.
+    """
+    semaphore = asyncio.Semaphore(concurrency)
+    async with httpx.AsyncClient(transport=transport) as client:
+        tasks = [_fetch(client, url, semaphore) for url in urls]
+        return await asyncio.gather(*tasks)
+
+
+if __name__ == "__main__":
+    targets = [f"https://books.toscrape.com/catalogue/page-{i}.html" for i in range(1, 4)]
+    pages = asyncio.run(scrape(targets))
+    print("fetched:", sum(page is not None for page in pages))

pws_examples/extraction.py

@@ -0,0 +1,23 @@
+"""Text extraction with regular expressions.
+
+Companion code for:
+- https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/extracting-data-with-regular-expressions/
+"""
+
+from __future__ import annotations
+
+import re
+
+EMAIL_RE = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+")
+PHONE_RE = re.compile(r"\b(?:\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b")
+
+
+def extract_contacts(text: str) -> dict[str, list[str]]:
+    """Return de-duplicated, sorted emails and phone numbers found in free text."""
+    emails = sorted(set(EMAIL_RE.findall(text)))
+    phones = sorted(set(PHONE_RE.findall(text)))
+    return {"emails": emails, "phones": phones}
+
+
+if __name__ == "__main__":
+    print(extract_contacts("Reach us at info@python-web-scraping.com or +1 (555) 123-4567."))

pws_examples/http_client.py

@@ -0,0 +1,41 @@
+"""A resilient HTTP session with retries and exponential backoff.
+
+Companion code for:
+- https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/understanding-http-requests-and-responses/
+"""
+
+from __future__ import annotations
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+DEFAULT_USER_AGENT = (
+    "Mozilla/5.0 (compatible; pws-examples/1.0; +https://python-web-scraping.com)"
+)
+
+
+def build_session(retries: int = 3, backoff_factor: float = 1.0) -> requests.Session:
+    """Create a ``requests.Session`` that retries transient failures politely.
+
+    Retries ``429`` and ``5xx`` responses on idempotent methods with exponential
+    backoff, and sets a transparent, identifying User-Agent.
+    """
+    session = requests.Session()
+    retry = Retry(
+        total=retries,
+        backoff_factor=backoff_factor,
+        status_forcelist=[429, 500, 502, 503, 504],
+        allowed_methods=["GET", "HEAD"],
+    )
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    session.headers.update({"User-Agent": DEFAULT_USER_AGENT})
+    return session
+
+
+if __name__ == "__main__":
+    session = build_session()
+    adapter = session.get_adapter("https://example.com")
+    print("Session ready, total retries:", adapter.max_retries.total)

pws_examples/parsing.py

@@ -0,0 +1,50 @@
+"""HTML parsing with BeautifulSoup.
+
+Companion code for:
+- https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/parsing-html-with-beautifulsoup/
+- https://python-web-scraping.com/the-complete-guide-to-python-web-scraping/understanding-http-requests-and-responses/step-by-step-guide-to-extracting-tables-from-html/
+"""
+
+from __future__ import annotations
+
+from bs4 import BeautifulSoup
+
+
+def parse_products(html: str) -> list[dict]:
+    """Extract product cards (title + price) from a listing page's HTML."""
+    soup = BeautifulSoup(html, "lxml")
+    products: list[dict] = []
+    for card in soup.select("article.product_pod"):
+        title = card.select_one("h3 a")
+        price = card.select_one("p.price_color")
+        if title and price:
+            products.append(
+                {
+                    "title": title.get("title") or title.get_text(strip=True),
+                    "price": price.get_text(strip=True),
+                }
+            )
+    return products
+
+
+def extract_table(html: str) -> list[dict]:
+    """Convert the first ``<table>`` into a list of row dicts keyed by header."""
+    soup = BeautifulSoup(html, "lxml")
+    table = soup.find("table")
+    if table is None:
+        return []
+    headers = [th.get_text(strip=True) for th in table.select("thead th")]
+    rows: list[dict] = []
+    for tr in table.select("tbody tr"):
+        cells = [td.get_text(strip=True) for td in tr.find_all("td")]
+        if cells:
+            rows.append(dict(zip(headers, cells, strict=False)))
+    return rows
+
+
+if __name__ == "__main__":
+    demo = (
+        '<article class="product_pod"><h3><a title="Clean Code">Clean Code</a></h3>'
+        '<p class="price_color">£42.00</p></article>'
+    )
+    print(parse_products(demo))

pws_examples/storage.py

@@ -0,0 +1,59 @@
+"""Validate scraped records and store them, de-duplicated, in SQLite.
+
+Companion code for:
+- https://python-web-scraping.com/scaling-python-web-scrapers/storing-and-exporting-scraped-data/
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from pydantic import BaseModel, ValidationError, field_validator
+
+
+class Product(BaseModel):
+    """A validated product record. ``price`` is coerced from strings like '£42.00'."""
+
+    title: str
+    price: float
+    in_stock: bool = True
+
+    @field_validator("price", mode="before")
+    @classmethod
+    def _clean_price(cls, value: object) -> float:
+        return float(str(value).replace("£", "").replace("$", "").strip())
+
+
+def validate(rows: list[dict]) -> list[Product]:
+    """Validate raw dict rows into ``Product`` models, skipping malformed ones."""
+    products: list[Product] = []
+    for row in rows:
+        try:
+            products.append(Product(**row))
+        except ValidationError:
+            continue
+    return products
+
+
+def store_sqlite(products: list[Product], conn: sqlite3.Connection) -> int:
+    """Create the table if needed and insert, de-duplicating by title.
+
+    Returns the number of rows actually inserted.
+    """
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS products ("
+        "title TEXT UNIQUE, price REAL, in_stock INTEGER)"
+    )
+    before = conn.total_changes
+    conn.executemany(
+        "INSERT OR IGNORE INTO products (title, price, in_stock) VALUES (?, ?, ?)",
+        [(p.title, p.price, int(p.in_stock)) for p in products],
+    )
+    conn.commit()
+    return conn.total_changes - before
+
+
+if __name__ == "__main__":
+    connection = sqlite3.connect(":memory:")
+    records = validate([{"title": "Clean Code", "price": "£42.00"}, {"title": "no price"}])
+    print("stored:", store_sqlite(records, connection))