Building a Resilient Web Scraping & Alert Pipeline in Python

Most web scraping tutorials show you how to fetch a static page with requests and print a price to the terminal. While fine for toy demos, that approach quickly falls apart in production: servers throttle requests, network connections drop, and website layouts change unexpectedly.

In this tutorial, we will build a production-ready, asynchronous price-monitoring pipeline in Python.

By the end of this guide, you will have a pipeline that:

Fetches multiple target URLs concurrently using aiohttp.
Parses product data reliably with BeautifulSoup4.
Handles retries with exponential backoff.
Dispatches formatted alert notifications to a Discord or Slack webhook when target thresholds are hit.

Architecture Overview

Here is how data flows through our pipeline:

[Target URLs] │ ▼ [Async HTTP Fetcher] ──(Backoff / Retries)──► [Network Resilience] │ ▼ [Parser & Validator] ──(Extracts price & title safely) │ ▼ [Threshold Engine] ───(Compares current vs. desired price) │ ▼ [Discord Webhook Alert]

Prerequisites

Before starting, ensure you have:

Python 3.9+ installed.
A basic understanding of asynchronous Python (async/await).
A Discord channel with an active Webhook URL (or a dummy URL for testing).

Create and activate a virtual environment, then install the required dependencies:

python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate pip install aiohttp beautifulsoup4 pydantic

Step 1: Define the Data Models with Pydantic

Strong typing prevents silent failures when parsing scraped HTML. Let's create models.py to define our data structures.

# models.py from pydantic import BaseModel, HttpUrl, Field from typing import Optional class TargetProduct(BaseModel): name: str url: HttpUrl target_price: float = Field(gt=0, description="Alert if price drops below this value") class ScrapedResult(BaseModel): name: str url: str current_price: Optional[float] = None in_stock: bool = False error: Optional[str] = None

Using Pydantic guarantees that invalid target configurations fail loudly at startup rather than mid-execution.

Step 2: Build the Asynchronous Fetcher with Retries

Scrapers frequently encounter temporary rate limits (429 Too Many Requests) or connection timeouts. Instead of failing immediately, we implement exponential backoff.

Create fetcher.py:

# fetcher.py import asyncio import logging import aiohttp from typing import Optional logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") DEFAULT_HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept-Language": "en-US,en;q=0.9", } async def fetch_html_with_retry( session: aiohttp.ClientSession, url: str, max_retries: int = 3, base_delay: float = 1.5, ) -> Optional[str]: """ Fetches raw HTML from a URL with exponential backoff on transient errors. """ for attempt in range(1, max_retries + 1): try: async with session.get(url, headers=DEFAULT_HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: return await response.text() elif response.status in (429, 500, 502, 503, 504): delay = base_delay * (2 ** (attempt - 1)) logging.warning(f"HTTP {response.status} for {url}. Retrying in {delay:.1f}s (Attempt {attempt}/{max_retries})...") await asyncio.sleep(delay) else: logging.error(f"Unrecoverable HTTP {response.status} for {url}") return None except (aiohttp.ClientError, asyncio.TimeoutError) as err: delay = base_delay * (2 ** (attempt - 1)) logging.warning(f"Network error on {url}: {err}. Retrying in {delay:.1f}s...") await asyncio.sleep(delay) logging.error(f"Exceeded max retries for {url}") return None

Why this matters:

ClientTimeout(total=10): Prevents requests from hanging indefinitely.
Exponential backoff (2 ** (attempt - 1)): Gives the target server breathing room instead of slamming it with immediate retries.
Custom User-Agent: Emulates a standard browser header to avoid naive bot blockers.

Step 3: Implement Resilient HTML Parsing

Websites change selectors frequently. We isolate our extraction logic inside parser.py and sanitize price strings into clean floating-point numbers.

# parser.py import re from typing import Optional from bs4 import BeautifulSoup from models import TargetProduct, ScrapedResult def parse_product_html(html: str, target: TargetProduct) -> ScrapedResult: """ Parses product details safely from HTML using multi-selector fallbacks. """ soup = BeautifulSoup(html, "html.parser") # Common selector patterns for prices across modern storefronts  price_selectors = [ ".price", "#priceblock_ourprice", ".a-price-whole", "[data-testid='product-price']", ".current-price" ] extracted_price: Optional[float] = None for selector in price_selectors: element = soup.select_one(selector) if element and element.get_text(strip=True): raw_text = element.get_text(strip=True) # Match numbers like $129.99 or 1,299.00  cleaned_match = re.search(r"(\d+[\d,]*\.?\d*)", raw_text.replace(",", "")) if cleaned_match: extracted_price = float(cleaned_match.group(1)) break # Check for Out of Stock indicators  stock_text = soup.get_text().lower() out_of_stock = any(phrase in stock_text for phrase in ["out of stock", "currently unavailable", "sold out"]) return ScrapedResult( name=target.name, url=str(target.url), current_price=extracted_price, in_stock=not out_of_stock, error="Price not found" if extracted_price is None else None )

Step 4: Dispatching Alerts via Discord Webhook

When a price drops below our target, we notify the user immediately via an embedded message.

Create notifier.py:

# notifier.py import aiohttp import logging from models import ScrapedResult, TargetProduct async def send_discord_alert( session: aiohttp.ClientSession, webhook_url: str, product: TargetProduct, result: ScrapedResult ) -> bool: """ Sends an alert card to a Discord webhook channel. """ payload = { "embeds": [ { "title": f"🚨 Price Drop Alert: {product.name}", "url": str(product.url), "color": 3066993, # Green  "fields": [ {"name": "Current Price", "value": f"${result.current_price:.2f}", "inline": True}, {"name": "Target Price", "value": f"${product.target_price:.2f}", "inline": True}, {"name": "Status", "value": "In Stock" if result.in_stock else "Out of Stock", "inline": False}, ], "footer": {"text": "Python Automation Pipeline"}, } ] } try: async with session.post(webhook_url, json=payload) as resp: return resp.status in (200, 204) except Exception as exc: logging.error(f"Failed to deliver webhook alert: {exc}") return False

Step 5: Putting It All Together

Create the orchestrator in main.py using asyncio.gather to scrape multiple targets concurrently:

# main.py import asyncio import os import aiohttp from models import TargetProduct from fetcher import fetch_html_with_retry from parser import parse_product_html from notifier import send_discord_alert # Replace with your actual Discord Webhook or an environment variable DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "https://discord.com/api/webhooks/YOUR_KEY") TARGET_WATCHLIST = [ TargetProduct( name="Mechanical Keyboard", url="https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html", # Safe sandbox url  target_price=60.00 ), TargetProduct( name="Noise Cancelling Headphones", url="https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html", target_price=55.00 ), ] async def monitor_target(session: aiohttp.ClientSession, target: TargetProduct): html = await fetch_html_with_retry(session, str(target.url)) if not html: print(f"[-] Could not retrieve page for: {target.name}") return result = parse_product_html(html, target) print(f"[+] Scraped {result.name}: Current=${result.current_price} | In Stock={result.in_stock}") # Trigger alert if criteria is met  if result.current_price and result.current_price <= target.target_price and result.in_stock: print(f"[!] Target hit for {target.name}! Sending alert...") await send_discord_alert(session, DISCORD_WEBHOOK_URL, target, result) async def main(): async with aiohttp.ClientSession() as session: tasks = [monitor_target(session, target) for target in TARGET_WATCHLIST] await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main())

Production Best Practices

When deploying this pipeline as a scheduled job (e.g., via GitHub Actions or a cron daemon):

Rotate IP Addresses: For large-scale scrapers (hundreds of URLs), route requests through residential proxies.
Respect robots.txt: Always check the site's scraping policies and avoid hitting internal endpoints too aggressively.
Persist State: Store previous prices in SQLite or Redis to calculate historical price trends.