Python Scraping Patterns: 15 Real Problems, 15 Working Solutions

Every scraping project hits the same walls. You write what looks like correct code, point it at a real website, and something doesn't work. The element isn't there. The data is missing. The response is empty. The selector returns None.

This is a reference guide. Fifteen patterns you will hit. Fifteen solutions that work. Bookmark it, come back when you're stuck.

Pattern 1: The Element You Need Doesn't Exist in BeautifulSoup

The problem: soup.select_one(".price") returns None even though the element is clearly visible in your browser.

Why it happens: The page uses JavaScript to render the content. BeautifulSoup parses the initial HTML response from the server — before JavaScript runs. Any element injected by JavaScript after page load is invisible to BeautifulSoup.

The signal: Open the page, right-click the element, click "View Page Source." If the element isn't in the raw source, it's JavaScript-rendered.

Solution A — Use Playwright to get the rendered HTML:

from playwright.sync_api import sync_playwright
from bs4 import BeautifulSoup

def get_rendered_html(url: str) -> str:
    """Fetch fully rendered HTML after JavaScript executes."""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page    = browser.new_page()
        page.goto(url, wait_until="networkidle")  # Wait for JS to finish
        html = page.content()   # This is the post-JS DOM
        browser.close()
    return html

html = get_rendered_html("https://js-heavy-site.com/products")
soup = BeautifulSoup(html, "lxml")
price = soup.select_one(".price")   # Now it finds it

Solution B — Find the XHR/fetch call that loads the data (faster, cheaper):

1. Open DevTools → Network tab → filter by XHR/Fetch

2. Reload the page

3. Find the API call that returns JSON with your data

4. Call that endpoint directly — no browser needed

import httpx

# Instead of scraping the rendered page, call the API the page uses
r = httpx.get(
    "https://site.com/api/products?category=electronics",
    headers={"User-Agent": "Mozilla/5.0 Chrome/120"}
)
data = r.json()   # Clean structured data, no parsing needed

When to use B over A: Always try B first. It's 10–50x faster and requires no browser.

Pattern 2: The Data Is in a <script> Tag as JSON

The problem: The data you want is on the page but not in HTML elements — it's embedded in a JavaScript variable inside a <script> tag.

Why it happens: Modern frameworks (Next.js, Nuxt, SvelteKit) embed server-side data as JSON in <script id="__NEXT_DATA__"> or similar tags for client-side hydration.

Solution:

import httpx
import json
import re
from bs4 import BeautifulSoup

def extract_next_data(url: str) -> dict:
    """Extract data from Next.js's __NEXT_DATA__ script tag."""
    r    = httpx.get(url, headers={"User-Agent": "Mozilla/5.0 Chrome/120"})
    soup = BeautifulSoup(r.text, "lxml")

    # Method 1: __NEXT_DATA__ (Next.js)
    script = soup.find("script", {"id": "__NEXT_DATA__"})
    if script:
        return json.loads(script.string)

    # Method 2: Variable assignment pattern — var data = {...}
    for script in soup.find_all("script"):
        text = script.string or ""
        # Match: var productData = {...} or window.__data = {...}
        match = re.search(
            r'(?:var\s+\w+|window\.\w+)\s*=\s*(\{.+?\});',
            text,
            re.DOTALL
        )
        if match:
            try:
                return json.loads(match.group(1))
            except json.JSONDecodeError:
                continue

    return {}

data = extract_next_data("https://nextjs-site.com/product/123")
# Navigate the JSON tree to your data
product = data.get("props", {}).get("pageProps", {}).get("product", {})
print(product.get("price"))

Pattern 3: Infinite Scroll Pages

The problem: A page loads 20 items. Scrolling loads 20 more. You need all of them. BeautifulSoup only sees the initial 20.

Why it happens: The site uses Intersection Observer API or scroll event listeners to trigger additional XHR requests as you scroll.

Solution A — Intercept the pagination API (preferred):

import httpx
import asyncio

# Infinite scroll almost always calls an API endpoint with page/offset params
# Find it in DevTools → Network → XHR → scroll the page → watch which call fires

async def scrape_all_via_api(base_url: str, limit: int = 20) -> list[dict]:
    all_items = []
    offset    = 0

    async with httpx.AsyncClient() as client:
        while True:
            r    = await client.get(base_url, params={"limit": limit, "offset": offset})
            data = r.json()

            items = data.get("items") or data.get("results") or data.get("data", [])
            if not items:
                break

            all_items.extend(items)
            print(f"  Fetched {len(all_items)} total")

            if len(items) < limit:   # Last page
                break
            offset += limit

    return all_items

items = asyncio.run(scrape_all_via_api("https://api.site.com/feed"))

Solution B — Playwright scroll automation:

from playwright.sync_api import sync_playwright
import time

def scrape_infinite_scroll(url: str, max_scrolls: int = 20) -> list[str]:
    """Scroll down, wait for new content, collect all visible items."""
    with sync_playwright() as p:
        browser  = p.chromium.launch(headless=True)
        page     = browser.new_page()
        page.goto(url, wait_until="networkidle")

        seen_items = set()
        all_text   = []

        for _ in range(max_scrolls):
            # Collect current items
            items = page.locator(".item-card").all()
            for item in items:
                text = item.inner_text()
                if text not in seen_items:
                    seen_items.add(text)
                    all_text.append(text)

            # Scroll to bottom
            page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            time.sleep(2)   # Wait for new items to load

            # Check if height changed (new content loaded)
            new_height = page.evaluate("document.body.scrollHeight")

        browser.close()
    return all_text

Pattern 4: Login Walls — Scraping Content That Requires Authentication

The problem: The content you need is behind a login. Navigating to the page redirects to a login form.

Solution — Save and reuse session cookies:

import httpx
import json
from pathlib import Path

COOKIES_FILE = Path("session.json")

def login_and_save(login_url: str, username: str, password: str,
                   username_field: str = "email",
                   password_field: str = "password"):
    """
    Log in via form POST and save session cookies.
    Works for most standard HTML login forms.
    """
    with httpx.Client(follow_redirects=True) as client:
        # First GET the login page (get CSRF token if present)
        login_page = client.get(login_url)
        soup       = BeautifulSoup(login_page.text, "lxml")

        # Extract CSRF token if present
        csrf_input = soup.find("input", {"name": ["csrf_token", "_token",
                                                   "authenticity_token"]})
        csrf_value = csrf_input.get("value", "") if csrf_input else ""

        # POST login credentials
        client.post(login_url, data={
            username_field: username,
            password_field: password,
            "csrf_token":   csrf_value,   # Include CSRF if found
        })

        # Save cookies
        cookies_dict = dict(client.cookies)
        with open(COOKIES_FILE, "w") as f:
            json.dump(cookies_dict, f)
        print(f"Saved {len(cookies_dict)} session cookies")

def scrape_with_session(url: str) -> str:
    """Make requests using a saved login session."""
    if not COOKIES_FILE.exists():
        raise RuntimeError("No session found. Run login_and_save() first.")

    with open(COOKIES_FILE) as f:
        cookies = json.load(f)

    r = httpx.get(url, cookies=cookies,
                  headers={"User-Agent": "Mozilla/5.0 Chrome/120"},
                  follow_redirects=True)

    # Detect if we got redirected to login page (session expired)
    if "login" in r.url.path.lower() or "sign-in" in r.url.path.lower():
        raise RuntimeError("Session expired. Re-login required.")

    return r.text

Pattern 5: Rate Limited — Getting 429 Too Many Requests

The problem: Your scraper works for the first 20–50 requests then starts getting 429 Too Many Requests.

Why it happens: The server tracks request frequency per IP and throttles clients that exceed a threshold.

Solution — Adaptive backoff with jitter:

import asyncio
import random
import httpx
from typing import Callable

class AdaptiveRateLimiter:
    """
    Automatically adjusts request delay based on response codes.
    Slows down when throttled, speeds up when succeeding.
    """

    def __init__(self, min_delay: float = 0.5, max_delay: float = 30.0):
        self.delay     = min_delay
        self.min_delay = min_delay
        self.max_delay = max_delay

    async def wait(self):
        # Add jitter: ±20% of current delay
        jitter = self.delay * random.uniform(0.8, 1.2)
        await asyncio.sleep(jitter)

    def on_success(self):
        # Gradually speed up when things work
        self.delay = max(self.min_delay, self.delay * 0.95)

    def on_rate_limit(self, retry_after: int = None):
        # Back off hard when throttled
        self.delay = min(self.max_delay, self.delay * 2.0)
        print(f"  Rate limited. New delay: {self.delay:.1f}s")

    def on_error(self):
        # Moderate backoff on errors
        self.delay = min(self.max_delay, self.delay * 1.5)

async def scrape_with_rate_limiting(urls: list[str]) -> list[dict]:
    limiter = AdaptiveRateLimiter(min_delay=1.0)
    results = []

    async with httpx.AsyncClient() as client:
        for url in urls:
            await limiter.wait()

            try:
                r = await client.get(url, timeout=15)

                if r.status_code == 429:
                    retry_after = int(r.headers.get("Retry-After", 30))
                    limiter.on_rate_limit(retry_after)
                    await asyncio.sleep(retry_after)
                    continue   # Retry this URL next iteration

                elif r.status_code == 200:
                    limiter.on_success()
                    results.append({"url": url, "html": r.text})

                else:
                    limiter.on_error()

            except httpx.TimeoutException:
                limiter.on_error()
                print(f"  Timeout: {url}")

    return results

Pattern 6: Paginated Results with No "Next" Link

The problem: A site has pagination but no visible "next page" link — just a page number in the URL.

Solution — URL pattern detection:

import httpx
import asyncio
import re
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse

def detect_pagination_pattern(url: str, html: str) -> dict | None:
    """
    Auto-detect pagination URL pattern from a page.
    Returns dict with info needed to generate all page URLs.
    """
    soup   = BeautifulSoup(html, "lxml")
    parsed = urlparse(url)

    # Strategy 1: Look for numbered page links
    page_links = soup.select("a[href]")
    page_numbers = set()
    for link in page_links:
        href = link.get("href", "")
        # Match ?page=N or /page/N or &p=N patterns
        match = re.search(r'[?&/](?:page|p|pg)=?/?(\d+)', href)
        if match:
            page_numbers.add(int(match.group(1)))

    if page_numbers:
        max_page = max(page_numbers)
        return {"type": "query_param", "max_page": max_page}

    # Strategy 2: Look for total count + items per page
    count_el = soup.select_one(".total-results, .result-count, [data-total]")
    if count_el:
        count_text = count_el.get_text()
        count_match = re.search(r"([\d,]+)", count_text)
        if count_match:
            total = int(count_match.group(1).replace(",", ""))
            return {"type": "offset", "total": total}

    return None

async def scrape_all_pages(base_url: str, page_param: str = "page",
                            items_per_page: int = 20) -> list[str]:
    """Scrape all pages of a paginated result set."""
    all_htmls = []

    # Fetch first page and detect pagination
    async with httpx.AsyncClient() as client:
        r    = await client.get(base_url)
        all_htmls.append(r.text)
        info = detect_pagination_pattern(base_url, r.text)

        if not info:
            return all_htmls   # Only one page

        max_page = info.get("max_page") or (info.get("total", 0) // items_per_page + 1)

        # Build and fetch all remaining pages
        parsed = urlparse(base_url)
        for page_num in range(2, max_page + 1):
            params = parse_qs(parsed.query)
            params[page_param] = [str(page_num)]
            new_query = urlencode(params, doseq=True)
            page_url  = urlunparse(parsed._replace(query=new_query))

            r = await client.get(page_url)
            all_htmls.append(r.text)
            await asyncio.sleep(1.0)
            print(f"  Page {page_num}/{max_page}")

    return all_htmls

Pattern 7: Downloading Files (PDFs, CSVs, Images) at Scale

The problem: You need to download thousands of files linked from scraped pages — annual reports, product images, data exports.

Solution — Async parallel downloader with resume support:

import asyncio
import httpx
import hashlib
from pathlib import Path
from tqdm.asyncio import tqdm

DOWNLOAD_DIR = Path("downloads")
DOWNLOAD_DIR.mkdir(exist_ok=True)
SEMAPHORE    = asyncio.Semaphore(10)   # 10 concurrent downloads

def local_path(url: str, extension: str = None) -> Path:
    """Generate a stable local filename from a URL."""
    url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
    ext      = extension or Path(url.split("?")[0]).suffix or ".bin"
    return DOWNLOAD_DIR / f"{url_hash}{ext}"

async def download_file(client: httpx.AsyncClient, url: str,
                        force: bool = False) -> Path | None:
    """Download a single file. Skips if already downloaded (resume support)."""
    dest = local_path(url)

    if dest.exists() and not force:
        return dest   # Already downloaded — skip

    async with SEMAPHORE:
        try:
            async with client.stream("GET", url, timeout=30) as r:
                r.raise_for_status()
                # Use content-type to get the right extension
                content_type = r.headers.get("content-type", "")
                if ".pdf" not in str(dest) and "pdf" in content_type:
                    dest = local_path(url, ".pdf")

                with open(dest, "wb") as f:
                    async for chunk in r.aiter_bytes(8192):
                        f.write(chunk)
            return dest
        except Exception as e:
            print(f"  Failed {url}: {e}")
            return None

async def batch_download(urls: list[str]) -> list[Path]:
    """Download all URLs concurrently with progress bar."""
    headers = {"User-Agent": "Mozilla/5.0 Chrome/120"}
    results = []

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
        tasks   = [download_file(client, url) for url in urls]
        results = await tqdm.gather(*tasks, desc="Downloading", unit="file")

    successful = [r for r in results if r]
    print(f"\nDownloaded {len(successful)}/{len(urls)} files to {DOWNLOAD_DIR}")
    return successful

# Example: download all PDFs from a page
pdf_urls = [
    "https://example.com/report-2024.pdf",
    "https://example.com/report-2023.pdf",
    "https://example.com/annual-2022.pdf",
]

paths = asyncio.run(batch_download(pdf_urls))

Pattern 8: Scraping Tables with Merged Cells

The problem: pd.read_html() works on simple tables but returns garbled data when rows have colspan or rowspan attributes.

Solution — Manual cell expansion:

def parse_table_with_spans(table_element) -> list[list[str]]:
    """
    Parse an HTML table that contains rowspan/colspan cells correctly.
    Returns a rectangular 2D list of strings.
    """
    # First pass: determine table dimensions
    rows     = table_element.find_all("tr")
    num_rows = len(rows)
    num_cols = 0
    for row in rows:
        cols = sum(
            int(cell.get("colspan", 1))
            for cell in row.find_all(["td", "th"])
        )
        num_cols = max(num_cols, cols)

    # Build grid filled with None
    grid = [[None] * num_cols for _ in range(num_rows)]

    for row_idx, row in enumerate(rows):
        col_idx = 0
        for cell in row.find_all(["td", "th"]):
            # Skip cells already filled by a previous rowspan
            while col_idx < num_cols and grid[row_idx][col_idx] is not None:
                col_idx += 1

            rowspan = int(cell.get("rowspan", 1))
            colspan = int(cell.get("colspan", 1))
            value   = cell.get_text(strip=True)

            # Fill all cells this span covers
            for r in range(row_idx, min(row_idx + rowspan, num_rows)):
                for c in range(col_idx, min(col_idx + colspan, num_cols)):
                    grid[r][c] = value

            col_idx += colspan

    return grid

# Usage
import pandas as pd
from bs4 import BeautifulSoup

html = """<table>
  <tr><th colspan="2">Product</th><th>Price</th></tr>
  <tr><td rowspan="2">Widget</td><td>Small</td><td>$9.99</td></tr>
  <tr><td>Large</td><td>$19.99</td></tr>
</table>"""

soup  = BeautifulSoup(html, "lxml")
table = soup.find("table")
grid  = parse_table_with_spans(table)
df    = pd.DataFrame(grid[1:], columns=grid[0])
print(df)
#   Product  Product  Price
# 0  Widget    Small  $9.99
# 1  Widget    Large $19.99

Pattern 9: Rotating User Agents Without Getting Detected

The problem: You rotate User-Agents but still get blocked because you use a Chrome UA with Firefox Accept headers — or vice versa.

Why it happens: Browsers send a characteristic set of headers that must be internally consistent. A Chrome User-Agent with Firefox headers is a red flag.

Solution — Consistent browser profiles:

import random

BROWSER_PROFILES = [
    {
        "User-Agent":        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "Accept":            "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
        "Accept-Language":   "en-US,en;q=0.9",
        "Accept-Encoding":   "gzip, deflate, br",
        "sec-ch-ua":         '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
        "sec-ch-ua-mobile":  "?0",
        "sec-ch-ua-platform":'"Windows"',
        "Sec-Fetch-Dest":    "document",
        "Sec-Fetch-Mode":    "navigate",
        "Sec-Fetch-Site":    "none",
        "Sec-Fetch-User":    "?1",
    },
    {
        "User-Agent":       "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36",
        "Accept":           "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
        "Accept-Language":  "en-US,en;q=0.9",
        "Accept-Encoding":  "gzip, deflate, br",
        "sec-ch-ua":        '"Not_A Brand";v="8", "Chromium";v="119", "Google Chrome";v="119"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform":'"macOS"',
        "Sec-Fetch-Dest":   "document",
        "Sec-Fetch-Mode":   "navigate",
        "Sec-Fetch-Site":   "none",
    },
    {
        "User-Agent":       "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
        "Accept":           "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
        "Accept-Language":  "en-US,en;q=0.5",
        "Accept-Encoding":  "gzip, deflate, br",
        "Sec-Fetch-Dest":   "document",
        "Sec-Fetch-Mode":   "navigate",
        "Sec-Fetch-Site":   "none",
        "Sec-Fetch-User":   "?1",
        "TE":               "trailers",
        # Note: no sec-ch-ua — Firefox doesn't send it
    },
]

def get_random_profile() -> dict:
    """Return a complete, internally consistent browser profile."""
    return random.choice(BROWSER_PROFILES).copy()

Pattern 10: Extracting Data Across Multiple Formats on the Same Page

The problem: A single page contains data in mixed formats — some in an HTML table, some in JSON inside a script tag, some in plain text paragraphs.

Solution — Multi-strategy extraction with fallback chain:

import re, json
from bs4 import BeautifulSoup

def extract_price(html: str) -> float | None:
    """
    Try multiple strategies to find a price, in order of reliability.
    Returns the first successful result.
    """
    soup = BeautifulSoup(html, "lxml")

    # Strategy 1: Structured data (most reliable) — JSON-LD schema.org
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string)
            if isinstance(data, list):
                data = data[0]
            price = (data.get("offers", {}).get("price")
                     or data.get("price"))
            if price:
                return float(str(price).replace(",", ""))
        except Exception:
            continue

    # Strategy 2: Meta tags (Open Graph / Twitter Cards)
    meta_price = soup.find("meta", property="product:price:amount")
    if meta_price:
        try:
            return float(meta_price.get("content", "").replace(",", ""))
        except ValueError:
            pass

    # Strategy 3: data-price attributes
    for el in soup.select("[data-price], [data-product-price]"):
        val = el.get("data-price") or el.get("data-product-price")
        if val:
            try:
                return float(str(val).replace(",", ""))
            except ValueError:
                continue

    # Strategy 4: CSS class patterns
    for selector in [".price", ".product-price", ".a-price-whole", "[itemprop='price']"]:
        el = soup.select_one(selector)
        if el:
            text = el.get_text(strip=True)
            match = re.search(r"[\d,]+\.?\d*", text.replace(",", ""))
            if match:
                try:
                    return float(match.group())
                except ValueError:
                    pass

    # Strategy 5: Regex on full page text (last resort)
    text = soup.get_text()
    match = re.search(r"(?:price|cost)[\s:$£€₹]*?([\d,]+\.?\d*)", text, re.IGNORECASE)
    if match:
        try:
            return float(match.group(1).replace(",", ""))
        except ValueError:
            pass

    return None   # Couldn't find a price

Pattern 11: Handling Cloudflare's Bot Protection Without a Proxy

The problem: curl_cffi with TLS impersonation works on most sites but still fails on Cloudflare-protected pages that use JavaScript challenges.

Solution — Detect the protection level and choose the right tool:

import httpx
from curl_cffi import requests as cffi_requests

def detect_protection_level(url: str) -> str:
    """
    Detect what level of bot protection a site uses.
    Returns: "none", "tls", "js_challenge", "captcha"
    """
    try:
        # First try with vanilla httpx
        r = httpx.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
        if r.status_code == 200 and len(r.text) > 1000:
            return "none"
        if r.status_code == 403:
            if "cloudflare" in r.text.lower():
                if "challenge" in r.text.lower() or "cf-browser-verification" in r.text:
                    return "js_challenge"
                return "tls"
            return "tls"
        if r.status_code == 429:
            return "rate_limit"
    except Exception:
        pass
    return "unknown"

async def fetch_adaptive(url: str) -> str | None:
    """
    Automatically pick the right fetching strategy based on protection level.
    """
    level = detect_protection_level(url)
    print(f"  Protection level: {level}")

    if level == "none":
        import httpx
        r = httpx.get(url)
        return r.text if r.status_code == 200 else None

    elif level == "tls":
        # TLS fingerprint impersonation fixes this
        r = cffi_requests.get(url, impersonate="chrome120", timeout=20)
        return r.text if r.status_code == 200 else None

    elif level == "js_challenge":
        # Need a full browser to solve JavaScript challenges
        from playwright.async_api import async_playwright
        from playwright_stealth import stealth_async
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page    = await browser.new_page()
            await stealth_async(page)
            await page.goto(url, wait_until="networkidle")
            html = await page.content()
            await browser.close()
        return html

    else:
        print(f"  Cannot automatically handle protection level: {level}")
        return None

Pattern 12: Storing Scraped Data Without Duplicates

The problem: You re-run your scraper and end up with duplicate rows in your database or CSV.

Solution — URL-based upsert pattern:

import sqlite3
import hashlib
from datetime import datetime, timezone

def create_dedup_store(db_path: str = "scrape_store.db"):
    """Create a SQLite store with built-in deduplication."""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS items (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            url_hash    TEXT UNIQUE NOT NULL,   -- MD5 of URL for fast lookup
            url         TEXT NOT NULL,
            title       TEXT,
            price       REAL,
            data_json   TEXT,                   -- Raw JSON for extra fields
            first_seen  TEXT DEFAULT (datetime('now')),
            last_updated TEXT DEFAULT (datetime('now'))
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_url_hash ON items(url_hash)
    """)
    conn.commit()
    return conn

def upsert_item(conn: sqlite3.Connection, url: str,
                title: str = None, price: float = None,
                extra: dict = None):
    """
    Insert a new item or update if the URL already exists.
    Tracks first_seen and last_updated separately.
    """
    import json
    url_hash = hashlib.md5(url.encode()).hexdigest()
    now      = datetime.now(timezone.utc).isoformat()

    conn.execute("""
        INSERT INTO items (url_hash, url, title, price, data_json, first_seen, last_updated)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(url_hash) DO UPDATE SET
            title        = excluded.title,
            price        = excluded.price,
            data_json    = excluded.data_json,
            last_updated = excluded.last_updated
    """, (url_hash, url, title, price,
          json.dumps(extra or {}), now, now))
    conn.commit()

# Usage
conn = create_dedup_store()
upsert_item(conn, "https://site.com/product/123", title="Widget", price=9.99)
upsert_item(conn, "https://site.com/product/123", title="Widget", price=8.99)  # Updates price, doesn't duplicate
upsert_item(conn, "https://site.com/product/456", title="Gadget", price=29.99)

import pandas as pd
df = pd.read_sql("SELECT * FROM items", conn)
print(df)
conn.close()

Pattern 13: Scraping Sites That Block Based on Geolocation

The problem: The site returns different content (or blocks you) based on your IP's country.

Solution — Geo-targeted requests:

from curl_cffi import requests as cffi_requests

# Country-specific proxy endpoints (from providers like Bright Data, Oxylabs)
COUNTRY_PROXIES = {
    "US": "http://user:[email protected]:8080",
    "UK": "http://user:[email protected]:8080",
    "IN": "http://user:[email protected]:8080",
    "DE": "http://user:[email protected]:8080",
}

def fetch_from_country(url: str, country: str = "US") -> dict:
    """
    Fetch a URL appearing to originate from a specific country.
    Useful for price comparison, geo-restricted content, localisation testing.
    """
    proxy = COUNTRY_PROXIES.get(country.upper())
    if not proxy:
        raise ValueError(f"No proxy configured for country: {country}")

    headers = {
        "User-Agent":      "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120",
        "Accept-Language": {
            "US": "en-US,en;q=0.9",
            "UK": "en-GB,en;q=0.9",
            "IN": "en-IN,en;q=0.9",
            "DE": "de-DE,de;q=0.9,en;q=0.8",
        }.get(country.upper(), "en-US,en;q=0.9")
    }

    r = cffi_requests.get(
        url,
        impersonate="chrome120",
        proxies={"https": proxy},
        headers=headers,
        timeout=20
    )
    return {"html": r.text, "status": r.status_code, "country": country}

# Compare prices across regions
def compare_regional_prices(product_url: str) -> dict:
    results = {}
    for country in ["US", "UK", "IN"]:
        response = fetch_from_country(product_url, country)
        soup     = BeautifulSoup(response["html"], "lxml")
        price_el = soup.select_one(".price, [data-price]")
        results[country] = price_el.get_text(strip=True) if price_el else "N/A"
    return results

Pattern 14: Detecting When a Scraper Has Gone Silent

The problem: Your scraper runs on a schedule but you don't know if it's actually collecting data. It might be blocked, broken, or returning empty results — silently.

Solution — Health check with alerting:

import sqlite3
import smtplib
from datetime import datetime, timezone, timedelta
from email.mime.text import MIMEText

def health_check(db_path: str, alert_email: str, expected_hourly_rate: int = 100):
    """
    Check if the scraper is producing data at the expected rate.
    Sends an alert email if the rate drops below threshold.
    """
    conn = sqlite3.connect(db_path)

    # How many items were scraped in the last hour?
    one_hour_ago = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
    count = conn.execute(
        "SELECT COUNT(*) FROM items WHERE last_updated > ?", (one_hour_ago,)
    ).fetchone()[0]
    conn.close()

    status = {
        "timestamp":      datetime.now(timezone.utc).isoformat(),
        "items_last_hour":count,
        "expected_rate":  expected_hourly_rate,
        "health":         "OK" if count >= expected_hourly_rate * 0.5 else "DEGRADED",
    }

    print(f"Health: {status['health']} — {count} items in last hour "
          f"(expected ≥{expected_hourly_rate * 0.5:.0f})")

    if status["health"] != "OK":
        _send_alert(alert_email, status)

    return status

def _send_alert(to_email: str, status: dict):
    """Send a health alert email."""
    msg = MIMEText(
        f"Scraper health check FAILED\n\n"
        f"Timestamp:    {status['timestamp']}\n"
        f"Items/hour:   {status['items_last_hour']} (expected ≥{status['expected_rate'] * 0.5:.0f})\n"
        f"Status:       {status['health']}\n\n"
        f"Check the scraper logs immediately."
    )
    msg["Subject"] = f"⚠️ Scraper Alert: {status['health']}"
    msg["From"]    = "[email protected]"
    msg["To"]      = to_email

    try:
        with smtplib.SMTP("smtp.gmail.com", 587) as s:
            s.starttls()
            s.login("[email protected]", "app_password")
            s.send_message(msg)
        print(f"Alert sent to {to_email}")
    except Exception as e:
        print(f"Alert send failed: {e}")

# Schedule this to run every 30 minutes via cron:
# */30 * * * * python health_check.py
if __name__ == "__main__":
    health_check("scrape_store.db", "[email protected]", expected_hourly_rate=200)

Pattern 15: Scraping GraphQL APIs

The problem: The site uses GraphQL instead of REST. The Network tab shows POST requests to /graphql with JSON bodies.

Solution:

import httpx
import asyncio
import json

async def graphql_query(endpoint: str, query: str, variables: dict = None,
                        headers: dict = None) -> dict:
    """
    Execute a GraphQL query against any endpoint.
    Works for both public and authenticated GraphQL APIs.
    """
    default_headers = {
        "Content-Type":  "application/json",
        "User-Agent":    "Mozilla/5.0 Chrome/120",
        "Accept":        "application/json",
    }
    if headers:
        default_headers.update(headers)

    payload = {"query": query}
    if variables:
        payload["variables"] = variables

    async with httpx.AsyncClient(headers=default_headers) as client:
        r = await client.post(endpoint, json=payload, timeout=20)
        r.raise_for_status()
        data = r.json()

    if "errors" in data:
        raise Exception(f"GraphQL errors: {data['errors']}")

    return data.get("data", {})

async def scrape_shopify_storefront_graphql(store_domain: str,
                                             storefront_token: str) -> list[dict]:
    """
    Scrape a Shopify store's products using the official Storefront GraphQL API.
    Requires a Storefront API token (available free from Shopify partner dashboard).
    """
    endpoint = f"https://{store_domain}/api/2024-01/graphql.json"
    headers  = {"X-Shopify-Storefront-Access-Token": storefront_token}

    query = """
    query GetProducts(\(first: Int!, \)after: String) {
        products(first: \(first, after: \)after) {
            pageInfo {
                hasNextPage
                endCursor
            }
            edges {
                node {
                    id
                    title
                    handle
                    priceRange {
                        minVariantPrice { amount currencyCode }
                        maxVariantPrice { amount currencyCode }
                    }
                    availableForSale
                    tags
                    vendor
                }
            }
        }
    }
    """

    all_products = []
    cursor       = None

    while True:
        variables = {"first": 50, "after": cursor}
        data      = await graphql_query(endpoint, query, variables, headers)
        products  = data.get("products", {})
        edges     = products.get("edges", [])

        for edge in edges:
            node = edge["node"]
            all_products.append({
                "id":        node["id"],
                "title":     node["title"],
                "handle":    node["handle"],
                "min_price": node["priceRange"]["minVariantPrice"]["amount"],
                "currency":  node["priceRange"]["minVariantPrice"]["currencyCode"],
                "available": node["availableForSale"],
                "vendor":    node["vendor"],
            })

        page_info = products.get("pageInfo", {})
        if not page_info.get("hasNextPage"):
            break
        cursor = page_info.get("endCursor")
        print(f"  Fetched {len(all_products)} products...")

    return all_products

# Example usage
products = asyncio.run(scrape_shopify_storefront_graphql(
    store_domain     = "your-store.myshopify.com",
    storefront_token = "your-storefront-access-token",
))
print(f"Total products: {len(products)}")

Quick Reference

Published via ZyVOP — Write once in Markdown, auto-backup to GitHub, and syndicate to Dev.to, Medium & Hashnode in 1 click.