Blog

Keeping a website free of dead or mis‑directed URLs is a cornerstone of good SEO and user experience. Yet, as your site grows, manually scanning every page becomes impossible. That’s where a Python web broken link checker utility shines: it automates the hunt for 404 errors, redirects, and other link problems, giving you a clear, actionable report in minutes. In this guide we’ll explore why a link checker is essential, walk through building a robust Python script, and highlight best practices for integrating the tool into your development workflow.

Why a Broken Link Checker Matters for SEO and Users

Search engines treat broken links as a sign of neglect, which can hurt your rankings. Visitors, on the other hand, may abandon a page if they encounter a dead link, increasing bounce rates. A reliable link checker helps you:

• Maintain crawl efficiency – Search bots spend less time on error pages.
• Boost page authority – Healthy internal linking distributes link equity effectively.
• Improve accessibility – Screen readers and assistive technologies rely on functional links.
• Save time – Automated scans replace tedious manual checks.

Core Features of an Effective Python Link Checker

Before diving into code, let’s outline the capabilities your utility should have to be truly useful:

1. URL extraction – Parse HTML, Markdown, or plain text to find all href/src attributes.
2. Concurrent requests – Use asynchronous I/O or multithreading to speed up large scans.
3. Status classification – Distinguish between OK (200), redirects (301/302), client errors (404), and server errors (5xx).
4. Depth control – Limit crawling to a specific number of levels or domains.
5. Report generation – Output results in JSON, CSV, or a simple HTML summary.
6. Configurable timeout and retry logic – Handle flaky networks gracefully.

Step‑by‑Step: Building a Python Broken Link Checker

1. Set Up the Project Environment

Start with a clean virtual environment and install the required libraries:

python -m venv linkchecker-env
source linkchecker-env/bin/activate  # On Windows use `linkchecker-env\Scripts\activate`
pip install requests beautifulsoup4 aiohttp tqdm

2. Extract URLs from a Web Page

Using BeautifulSoup makes it easy to pull out href and src attributes:

import requests
from bs4 import BeautifulSoup

def get_links(url):
    response = requests.get(url, timeout=10)
    soup = BeautifulSoup(response.text, 'html.parser')
    links = set()
    for tag in soup.find_all(['a', 'img', 'script', 'link']):
        link = tag.get('href') or tag.get('src')
        if link:
            links.add(link)
    return links

3. Normalise and Filter URLs

Convert relative URLs to absolute ones and discard non‑HTTP links:

from urllib.parse import urljoin, urlparse

def normalise_links(base_url, links):
    normalised = set()
    for link in links:
        if link.startswith('#') or link.startswith('mailto:'):
            continue
        absolute = urljoin(base_url, link)
        parsed = urlparse(absolute)
        if parsed.scheme in ('http', 'https'):
            normalised.add(absolute)
    return normalised

4. Asynchronous Link Validation

Scanning thousands of URLs sequentially is slow. aiohttp combined with asyncio provides high performance:

import asyncio
import aiohttp
from tqdm import tqdm

async def fetch_status(session, url):
    try:
        async with session.head(url, allow_redirects=True, timeout=10) as resp:
            return url, resp.status
    except Exception:
        return url, None

async def check_links(urls):
    results = {}
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_status(session, url) for url in urls]
        for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)):
            url, status = await f
            results[url] = status
    return results

5. Orchestrating the Crawl

Combine the functions above to crawl a site up to a defined depth:

def crawl(start_url, max_depth=2):
    visited = set()
    to_visit = [(start_url, 0)]
    all_links = set()

    while to_visit:
        current_url, depth = to_visit.pop(0)
        if current_url in visited or depth > max_depth:
            continue
        visited.add(current_url)

        try:
            raw_links = get_links(current_url)
            absolute_links = normalise_links(current_url, raw_links)
            all_links.update(absolute_links)

            if depth < max_depth:
                for link in absolute_links:
                    if urlparse(link).netloc == urlparse(start_url).netloc:
                        to_visit.append((link, depth + 1))
        except Exception as e:
            print(f"Error crawling {current_url}: {e}")

    return all_links

if __name__ == '__main__':
    import sys
    start = sys.argv[1] if len(sys.argv) > 1 else 'https://example.com'
    links = crawl(start, max_depth=2)
    results = asyncio.run(check_links(links))

    # Simple console report
    for url, status in results.items():
        status_text = status if status else 'Failed'
        print(f"{url} → {status_text}")

6. Generating an SEO‑Friendly Report

Export the findings to CSV for easy import into Google Sheets or an internal dashboard:

import csv

def export_csv(results, filename='link_report.csv'):
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['URL', 'Status'])
        for url, status in results.items():
            writer.writerow([url, status or 'Error'])
    print(f"Report saved to {filename}")

Best Practices for Ongoing Link Maintenance

Building the utility is only half the battle. To keep your site healthy over time, follow these operational guidelines:

• Integrate into CI/CD pipelines – Run the checker on every pull request to catch broken links before they go live.
• Schedule regular scans – A nightly or weekly cron job ensures new content doesn’t introduce errors.
• Prioritise critical pages – Focus on high‑traffic or conversion‑critical pages first.
• Automate fixes where possible – For known patterns (e.g., moved assets), script automatic URL updates.
• Monitor redirect chains – Long chains can dilute PageRank; keep them under three hops.

Advanced Enhancements You Might Consider

Once the basic checker is stable, you can extend its capabilities to match enterprise needs:

• Authentication support – Use session cookies or OAuth tokens to scan behind‑login areas.
• Headless browser integration – Leverage Playwright or Selenium to capture JavaScript‑generated links.
• Duplicate detection – Flag identical URLs that appear on multiple pages for potential consolidation.
• Link health scoring – Combine status codes with page load times to rank links by overall quality.
• Notification system – Send Slack, email, or webhook alerts when critical links break.

Common Pitfalls and How to Avoid Them

Even seasoned developers can run into snags. Keep an eye out for these issues:

• False positives from temporary 5xx errors – Implement retry logic with exponential backoff before marking a link as dead.
• Ignoring robots.txt – Respect crawl directives to avoid legal or performance problems.
• Overloading the target server – Throttle requests (e.g., 5‑10 concurrent calls) and honour Retry-After headers.
• Missing protocol-relative URLs – Ensure URLs that start with // are prefixed with the appropriate scheme.
• Hard‑coded timeouts – Use configurable settings so the utility adapts to different network environments.

Conclusion: Turn Link Checking into a Competitive Advantage

Dead links are more than just a nuisance—they’re a silent SEO penalty and a user‑experience risk. By deploying a custom Python web broken link checker utility, you gain granular control over how, when, and where links are validated, while keeping costs low and flexibility high. Whether you run a personal blog or a sprawling e‑commerce platform, automating link health checks empowers you to maintain a pristine web presence, boost search rankings, and deliver a smoother journey for every visitor.

In the era of information overload, staying updated with the latest news, blog posts, and podcasts can feel like a full‑time job. That’s where an RSS feed parser and aggregator built with Python shines – it automatically pulls content from multiple sources, normalizes the data, and delivers it straight to your application or inbox. In this guide we’ll explore why Python is a top choice for RSS handling, walk through the most popular libraries, and build a simple yet powerful aggregator that you can extend for personal projects or production‑grade services.

Why Choose Python for RSS Parsing?

Python’s clean syntax, extensive standard library, and vibrant ecosystem make it ideal for working with XML‑based formats like RSS and Atom. Here are a few reasons developers consistently pick Python for this task:

• Readability: Code is easy to understand and maintain, which is crucial when dealing with varied feed structures.
• Rich libraries: Modules such as feedparser, requests, and aiohttp handle the heavy lifting.
• Cross‑platform support: Write once, run on Windows, macOS, or Linux without modification.
• Scalability: Combine synchronous and asynchronous approaches to scrape hundreds of feeds efficiently.

Core Concepts Behind RSS Feeds

Before diving into code, it helps to understand the anatomy of an RSS feed. Most feeds are XML documents that follow one of two specifications:

• RSS 2.0: The classic format, featuring <channel>, <item>, <title>, <link>, and optional elements like <author> or <category>.
• Atom: A newer, more flexible standard that uses <feed> and <entry> tags.

Regardless of the version, the goal of a parser is to extract the same core fields – title, URL, publication date, and description – and present them in a uniform Python data structure.

Getting Started: Installing the Essentials

For most projects, the feedparser library is the go‑to solution. It abstracts away the differences between RSS and Atom, handling malformed XML gracefully. Pair it with requests for HTTP fetching, or aiohttp if you need asynchronous performance.

pip install feedparser requests aiohttp

Basic Synchronous Parser Example

The following snippet demonstrates a minimal synchronous RSS fetcher using requests and feedparser:

import requests
import feedparser

def fetch_feed(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()  # Raise an error for bad responses
    return feedparser.parse(response.content)

def print_titles(feed):
    for entry in feed.entries[:5]:  # Show first 5 items
        print(f"- {entry.title}")

if __name__ == "__main__":
    rss_url = "https://realpython.com/atom.xml"
    feed = fetch_feed(rss_url)
    print_titles(feed)

This script pulls the feed, parses it, and prints the titles of the latest five entries – a solid foundation for any aggregator.

Scaling Up with Asynchronous Requests

When aggregating dozens or hundreds of feeds, blocking I/O becomes a bottleneck. aiohttp combined with feedparser lets you fetch many feeds concurrently:

import asyncio
import aiohttp
import feedparser

async def fetch(session, url):
    async with session.get(url) as resp:
        content = await resp.read()
        return feedparser.parse(content)

async def gather_feeds(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    urls = [
        "https://realpython.com/atom.xml",
        "https://news.ycombinator.com/rss",
        "https://www.reddit.com/r/python/.rss",
    ]
    feeds = asyncio.run(gather_feeds(urls))
    for feed in feeds:
        print(f"Feed: {feed.feed.title}")
        for entry in feed.entries[:3]:
            print(f"  • {entry.title}")

The asynchronous version can retrieve dozens of feeds in a fraction of the time required by a synchronous loop.

Designing a Full‑Featured Aggregator

Beyond simple fetching, a production‑ready aggregator typically includes:

1. Feed storage: Persist raw XML or parsed entries in a database (SQLite, PostgreSQL, or NoSQL).
2. Duplicate detection: Avoid re‑processing the same article by tracking GUIDs or hashes.
3. Scheduling: Periodically refresh feeds using cron, APScheduler, or cloud‑based task queues.
4. Filtering & enrichment: Apply keyword filters, sentiment analysis, or add custom tags.
5. Output formats: Serve JSON APIs, generate HTML newsletters, or push to messaging platforms (Slack, Discord).

Sample Data Model with SQLite

Below is a lightweight schema that captures the essential fields for each article:

CREATE TABLE IF NOT EXISTS articles (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    feed_url TEXT NOT NULL,
    guid TEXT UNIQUE,
    title TEXT,
    link TEXT,
    published DATETIME,
    summary TEXT,
    fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

When inserting a new entry, the UNIQUE constraint on guid guarantees that duplicates are ignored.

Putting It All Together – A Minimal Aggregator Script

Here’s a compact example that combines async fetching, duplicate checks, and SQLite persistence. It can serve as a starter template for more complex systems.

import asyncio
import aiohttp
import feedparser
import sqlite3
from datetime import datetime

DB_PATH = "rss_aggregator.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""CREATE TABLE IF NOT EXISTS articles (
        id INTEGER PRIMARY KEY,
        feed_url TEXT,
        guid TEXT UNIQUE,
        title TEXT,
        link TEXT,
        published TEXT,
        summary TEXT,
        fetched_at TEXT
    )""")
    conn.commit()
    conn.close()

async def fetch(session, url):
    async with session.get(url, timeout=15) as resp:
        content = await resp.read()
        return url, feedparser.parse(content)

def save_entry(conn, feed_url, entry):
    guid = entry.get("id") or entry.get("guid") or entry.link
    published = entry.get("published", datetime.utcnow().isoformat())
    try:
        conn.execute(
            "INSERT INTO articles (feed_url, guid, title, link, published, summary) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (feed_url, guid, entry.title, entry.link, published, entry.summary)
        )
    except sqlite3.IntegrityError:
        # Duplicate GUID – skip insertion
        pass

async def aggregate(urls):
    init_db()
    conn = sqlite3.connect(DB_PATH)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        for feed_url, feed in await asyncio.gather(*tasks):
            for entry in feed.entries:
                save_entry(conn, feed_url, entry)
    conn.commit()
    conn.close()

if __name__ == "__main__":
    feed_list = [
        "https://realpython.com/atom.xml",
        "https://news.ycombinator.com/rss",
        "https://www.reddit.com/r/python/.rss",
        # Add more URLs as needed
    ]
    asyncio.run(aggregate(feed_list))
    print("Aggregation complete. Check the SQLite DB for stored articles.")

This script demonstrates the core workflow:

• Initialize the database.
• Fetch all feeds concurrently.
• Iterate over each entry, compute a reliable GUID, and insert it if it’s new.
• Close the connection cleanly.

Advanced Topics and Best Practices

Handling Authentication and Rate Limits

Some premium feeds require HTTP Basic Auth, OAuth, or API keys. aiohttp supports custom headers, while requests offers built‑in auth helpers. Remember to respect RFC 5005 guidelines and honor Retry-After headers to avoid being blocked.

Parsing Media Enclosures

Podcast feeds often embed audio files using the <enclosure> tag. feedparser exposes these via entry.enclosures. You can extract URLs, MIME types, and lengths to build a download queue or streaming service.

Testing Your Parser

Automated tests protect against breaking changes in feed formats. Use pytest together with sample XML fixtures:

def test_parse_realpython():
    with open("tests/fixtures/realpython.xml", "rb") as f:
        feed = feedparser.parse(f.read())
    assert feed.feed.title == "Real Python"
    assert len(feed.entries) > 0
    assert "Python" in feed.entries[0].title

Deploying at Scale

When

Looking to turn your Python skills into a real‑world product? A news aggregator web app is a perfect project that blends data fetching, web development, and user‑centric design—all while delivering a useful service that keeps visitors informed. In this guide we’ll walk you through every step of building a Python news aggregator web app, from choosing the right framework and pulling headlines via RSS or APIs, to storing articles, creating a sleek UI, and finally deploying the app for the world to see. Whether you’re a solo developer, a student, or part of a small team, this project will sharpen your Python chops and give you a showcase‑ready portfolio piece.

Why Build a News Aggregator with Python?

Benefits for Developers and Users

• Hands‑on experience with web scraping and APIs – learn to parse RSS feeds, consume JSON endpoints, and handle rate limits.
• Full‑stack development practice – combine Python back‑end logic with HTML/CSS front‑end frameworks like Bootstrap.
• Scalable architecture – design a modular codebase that can grow to include personalization, tagging, or machine‑learning recommendations.
• Portfolio boost – a live, functional web app demonstrates problem‑solving, data handling, and deployment skills to potential employers.

Core Features of a Modern News Aggregator

• Automatic fetching of articles from multiple sources (RSS, public APIs, or custom scrapers).
• Database storage with timestamps, categories, and source attribution.
• Responsive UI that lists headlines, shows summaries, and links to original articles.
• Search and filter capabilities by keyword, date, or category.
• User authentication for saved bookmarks or custom feeds (optional).
• Background scheduler to refresh content without manual intervention.

Choosing the Right Python Framework

Flask vs Django

Both Flask and Django are popular choices, but they serve different project scopes. Flask is lightweight, giving you full control over extensions and structure—ideal for a single‑purpose aggregator. Django offers a batteries‑included approach with built‑in ORM, admin panel, and authentication, which can speed up development if you plan to expand the app later.

For a starter project, Flask often wins on simplicity. Below is a minimal Flask route that renders the homepage.

from flask import Flask, render_template

app = Flask(__name__)

@app.route('/')
def home():
    # articles will be fetched from the database in a real app
    return render_template('index.html', articles=[])

if __name__ == '__main__':
    app.run(debug=True)

Setting Up the Project Environment

1. Install Python 3.11+ and create a virtual environment:

   python -m venv venv
   source venv/bin/activate   # Linux/macOS
   venv\Scripts\activate      # Windows

2. Install core dependencies:

   pip install Flask feedparser requests SQLAlchemy APScheduler

3. Initialize a Git repository and push to GitHub for version control.

   git init
   git add .
   git commit -m "Initial commit"
   git branch -M main
   git remote add origin https://github.com/yourusername/news-aggregator.git
   git push -u origin main

4. Create the following folder structure:

   news_aggregator/
   │
   ├─ app/
   │   ├─ __init__.py
   │   ├─ routes.py
   │   ├─ models.py
   │   └─ templates/
   │       └─ index.html
   ├─ migrations/
   ├─ static/
   │   └─ css/
   └─ run.py

Fetching News Sources – RSS and APIs

Most reputable news outlets expose RSS feeds, while others provide public JSON APIs (e.g., NewsAPI.org). The feedparser library makes RSS consumption painless.

import feedparser

def fetch_rss(url):
    feed = feedparser.parse(url)
    articles = []
    for entry in feed.entries:
        articles.append({
            'title': entry.title,
            'link': entry.link,
            'summary': entry.get('summary', ''),
            'published': entry.get('published', ''),
            'source': feed.feed.title
        })
    return articles

# Example usage
bbc_feed = 'http://feeds.bbci.co.uk/news/rss.xml'
bbc_articles = fetch_rss(bbc_feed)

For JSON APIs, use requests and handle authentication headers as needed.

Storing Articles in a Database

SQLAlchemy provides an ORM that works seamlessly with Flask or Django. Below is a simple model for storing articles.

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class Article(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(255), nullable=False)
    link = db.Column(db.String(512), unique=True, nullable=False)
    summary = db.Column(db.Text)
    published = db.Column(db.DateTime)
    source = db.Column(db.String(100))

    def __repr__(self):
        return f"<Article {self.title}>"

After defining the model, run migrations to create the table:

flask db init
flask db migrate -m "Create article table"
flask db upgrade

Building the Front‑End with Bootstrap

Bootstrap’s grid system lets you create a clean, responsive layout with minimal CSS. A typical article card might look like this:

<div class="card mb-3">
    <div class="card-body">
        <h5 class="card-title"><a href="{{ article.link }}" target="_blank">{{ article.title }}</a></h5>
        <p class="card-text">{{ article.summary|truncate(150) }}</p>
        <small class="text-muted">{{ article.source }} · {{ article.published.strftime('%b %d, %Y') }}</small>
    </div>
</div>

Combine this with a navigation bar, search form, and pagination to enhance usability.

Adding Search and Filtering

Implement a simple search endpoint that queries the database using SQLAlchemy’s filter method.

@app.route('/search')
def search():
    query = request.args.get('q', '')
    results = Article.query.filter(Article.title.ilike(f'%{query}%')).all()
    return render_template('search.html', articles=results, query=query)

For advanced filtering, consider adding columns for categories or tags and using JOIN queries.

Deploying the App to Production

• Choose a hosting platform – Heroku, Render, Fly.io, or a VPS with Docker.
• Configure Gunicorn (or uvicorn for FastAPI) as the WSGI server.
• Set environment variables for SECRET_KEY, database URL, and API keys.
• Schedule the news‑fetching job with APScheduler or a cloud cron service.
• Enable HTTPS with Let’s Encrypt or the platform’s built‑in SSL.

Example Procfile for Heroku:

web: gunicorn run:app
worker: python -m news_aggregator.scheduler

Best Practices and Common Pitfalls

• Respect source terms of service – avoid excessive polling; use ETag or Last-Modified headers when available.
• Cache API responses for a short period to reduce latency and API costs.
• Sanitize user‑generated input to prevent XSS attacks, especially when rendering article summaries.
• Implement graceful error handling for network timeouts or malformed feeds.
• Write unit tests for the fetcher, parser, and database layers to catch regressions early.

Conclusion

Building a Python news aggregator web app is more than just a coding exercise; it’s a comprehensive project that touches on data acquisition, back‑end architecture, front‑end design, and cloud deployment. By following the steps outlined above—selecting a framework, fetching RSS or API data, persisting articles, crafting a responsive UI, and finally launching to production—you’ll create

Imagine walking through an online store, spotting the perfect gadget, and then watching the price slip by a few dollars the next day. That fleeting discount can feel like a missed opportunity—unless you have a Python e‑commerce price drop alert tool working silently in the background. In this guide we’ll explore why price‑watching matters, break down the essential components of a robust alert system, and walk you through a step‑by‑step implementation that you can customize for any e‑commerce platform. By the end, you’ll have a ready‑to‑run script that automatically notifies you the moment a product’s price falls below your target, saving you time, money, and endless manual refreshing.

Why a Price Drop Alert Tool Is a Must‑Have for Shoppers and Sellers

Both consumers and online merchants benefit from real‑time price monitoring. Here’s why the market loves these tools:

• Instant savings: Catch discounts the moment they appear, avoiding the regret of a missed deal.
• Competitive edge: Sellers can dynamically adjust pricing based on competitor moves, keeping their offers attractive.
• Data‑driven decisions: Historical price trends reveal seasonal patterns, helping shoppers plan purchases and businesses forecast revenue.
• Automation: Replace manual page‑refresh habits with a reliable, hands‑free solution.

Core Components of a Python Price Drop Alert Tool

Building a reliable alert system boils down to three fundamental stages: gathering data, detecting changes, and delivering notifications.

1. Data Collection (Web Scraping or API Calls)

The first step is to retrieve the current price of a product. Depending on the site, you can either use a public API or scrape the HTML content. The goal is to extract a clean numeric value that can be compared over time.

2. Change Detection (Comparison Logic)

Once you have the latest price, compare it against a stored baseline (the last known price or a user‑defined threshold). If the new price is lower, flag the product as a “price drop.”

3. Notification System (Email, SMS, or Push)

A fast, reliable notification channel ensures the user knows about the discount instantly. Popular choices include SMTP email, Twilio SMS, or push services like Pushover.

Choosing the Right Python Libraries

Python’s ecosystem offers several mature libraries for each component. Selecting the appropriate tools can dramatically simplify development.

• Requests: Simple HTTP client for API calls and static pages.
• BeautifulSoup: Parses HTML and XML, perfect for extracting price tags from static sites.
• Selenium: Automates a real browser, handling JavaScript‑heavy pages.
• Scrapy: A full‑featured framework for large‑scale crawling and data pipelines.
• SQLAlchemy / SQLite: Stores historical prices in a lightweight relational database.
• SMTP (smtplib) or Twilio: Sends email or SMS alerts.

Step‑by‑Step Guide: Building the Alert Tool from Scratch

Step 1 – Set Up Your Development Environment

python -m venv price_alert_env
source price_alert_env/bin/activate  # On Windows use `price_alert_env\Scripts\activate`
pip install requests beautifulsoup4 sqlalchemy twilio

Using a virtual environment isolates dependencies and makes deployment easier.

Step 2 – Scrape Product Data

Below is a minimal example that fetches the price from a static product page using requests and BeautifulSoup. Adjust the CSS selector to match the target site.

import requests
from bs4 import BeautifulSoup

def fetch_price(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; PriceAlertBot/1.0)"
    }
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()

    soup = BeautifulSoup(response.text, "html.parser")
    # Example selector – replace with the actual element containing the price
    price_tag = soup.select_one(".product-price span")
    if not price_tag:
        raise ValueError("Price element not found")

    # Strip currency symbols and commas, then convert to float
    price_text = price_tag.get_text().strip().replace("$", "").replace(",", "")
    return float(price_text)

Step 3 – Store Historical Prices

Using SQLite via SQLAlchemy provides a simple, file‑based database that works anywhere Python runs.

from sqlalchemy import create_engine, Column, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime

Base = declarative_base()

class PriceRecord(Base):
    __tablename__ = "prices"
    url = Column(String, primary_key=True)
    price = Column(Float, nullable=False)
    last_checked = Column(DateTime, default=datetime.datetime.utcnow)

engine = create_engine("sqlite:///price_alert.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

Step 4 – Detect Price Drops

The logic compares the newly fetched price with the stored value. If the new price is lower, it triggers an alert and updates the record.

def check_price(url, target_price=None):
    session = Session()
    current_price = fetch_price(url)

    record = session.query(PriceRecord).filter_by(url=url).first()
    if record:
        if current_price < record.price:
            send_alert(url, current_price, record.price)
        record.price = current_price
        record.last_checked = datetime.datetime.utcnow()
    else:
        # First‑time entry
        record = PriceRecord(url=url, price=current_price)
        session.add(record)

    session.commit()
    session.close()
    return current_price

Step 5 – Send Alerts (Email Example)

Below is a lightweight email alert using Python’s built‑in smtplib. For production use, consider a service like SendGrid or Mailgun to improve deliverability.

import smtplib
from email.mime.text import MIMEText

SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = "your.email@gmail.com"
SMTP_PASS = "your_app_password"

def send_alert(url, new_price, old_price):
    subject = f"🔔 Price Drop Alert: Now ${new_price:.2f}"
    body = f"""\
The price of the product you’re watching has dropped!

🔗 Product: {url}
📉 Old Price: ${old_price:.2f}
📈 New Price: ${new_price:.2f}

Act fast before the deal disappears!
"""
    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = SMTP_USER
    msg["To"] = SMTP_USER  # Change to recipient list as needed

    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASS)
        server.send_message(msg)

Step 6 – Automate the Process

Schedule the script to run at regular intervals using cron (Linux/macOS) or Task Scheduler (Windows). For a cross‑platform solution, the schedule library can handle simple timing inside Python.

import schedule
import time

URLS = [
    "https://example.com/product/123",
    "https://shop.com/item/456",
]

def job():
    for url in URLS:
        try:
            price = check_price(url)
            print(f"Checked {url}: ${price:.2f}")
        except Exception as e:
            print(f"Error checking {url}: {e}")

schedule.every(30).minutes.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

Best Practices for Reliability and Legal Compliance

Scraping e‑commerce sites can be tricky. Follow these guidelines to keep your tool both effective and respectful.

• Respect robots.txt: Check the site’s crawling policy before extracting data.
• Rate limiting: Insert delays (e.g., time.sleep(2)) between requests to avoid overloading servers.
• Use rotating user‑agents: Mimic different browsers to reduce the chance of being blocked.
October 12, 2026

In the fast‑paced world of day trading and algorithmic investing, having access to real‑time stock prices can be the difference between a profitable trade and a missed opportunity. Python, with its rich ecosystem of libraries and straightforward syntax, makes it possible to build a reliable stock price scraper that fetches live data, processes it instantly, and integrates seamlessly with your trading strategies. In this guide we’ll walk through the entire process—from choosing the right data source and handling rate limits, to writing clean, maintainable code and deploying the scraper for continuous operation.

Why Build Your Own Real‑Time Stock Price Scraper?

While many platforms offer paid APIs, building a custom scraper provides several distinct advantages:

• Cost control: Avoid recurring subscription fees by leveraging free data endpoints.
• Flexibility: Tailor the data fields, frequency, and output format to match your exact workflow.
• Learning opportunity: Deepen your understanding of HTTP, WebSockets, and asynchronous programming in Python.
• Independence: Reduce reliance on third‑party services that may change terms or experience downtime.

Choosing the Right Data Source

Before writing any code, identify a reliable source that provides live quotes. Below are three popular options, each with its own trade‑offs:

1. Public REST APIs (e.g., Alpha Vantage, IEX Cloud)

These services expose endpoints that return JSON payloads. They are easy to use with the requests library but often impose strict rate limits (e.g., 5 requests/minute for free tiers).

2. WebSocket Streams (e.g., Finnhub, Polygon.io)

WebSockets push data to you as soon as a price update occurs, making them ideal for true real‑time needs. They require handling asynchronous connections, typically with websockets or asyncio.

3. HTML Scraping (e.g., Yahoo Finance, Google Finance)

When no API fits your budget, you can scrape the public web pages. This method demands careful parsing with BeautifulSoup and strict compliance with the site’s robots.txt and terms of service.

Setting Up the Development Environment

Start with a clean virtual environment to keep dependencies isolated:

python -m venv stock-scraper
source stock-scraper/bin/activate   # On Windows use `stock-scraper\Scripts\activate`
pip install requests websockets beautifulsoup4 lxml pandas

We’ll use pandas for data handling, requests for REST calls, websockets for live streams, and BeautifulSoup for any HTML parsing needs.

Implementing a Basic REST‑Based Scraper

If you’re just starting, a simple REST approach is the easiest way to fetch the latest price for a given ticker.

Step‑by‑Step Code Walkthrough

import time
import requests
import pandas as pd

API_KEY = 'YOUR_ALPHA_VANTAGE_KEY'
BASE_URL = 'https://www.alphavantage.co/query'

def get_latest_price(symbol: str) -> float:
    """Fetch the most recent closing price for *symbol* using Alpha Vantage."""
    params = {
        'function': 'TIME_SERIES_INTRADAY',
        'symbol': symbol,
        'interval': '1min',
        'apikey': API_KEY,
        'outputsize': 'compact'
    }
    response = requests.get(BASE_URL, params=params, timeout=10)
    data = response.json()

    # Guard against API errors or throttling
    if 'Time Series (1min)' not in data:
        raise RuntimeError(f'API error: {data.get("Note", "Unexpected response")}')

    # Extract the latest timestamp and its close price
    latest_timestamp = max(data['Time Series (1min)'])
    close_price = float(data['Time Series (1min)'][latest_timestamp]['4. close'])
    return close_price

if __name__ == '__main__':
    symbols = ['AAPL', 'MSFT', 'TSLA']
    while True:
        for sym in symbols:
            try:
                price = get_latest_price(sym)
                print(f'{sym}: ${price:.2f}')
            except Exception as e:
                print(f'Error fetching {sym}: {e}')
        time.sleep(60)  # Respect the free‑tier rate limit

This script pulls the most recent 1‑minute candle for each ticker, prints the closing price, and then sleeps for a minute to stay within Alpha Vantage’s free tier limits.

Scaling Up with WebSockets for True Real‑Time Data

When you need sub‑second updates, switch to a WebSocket provider. Below is a minimal example using Finnhub’s free real‑time stream.

WebSocket Example

import asyncio
import json
import websockets

FINNHUB_TOKEN = 'YOUR_FINNHUB_TOKEN'
WS_URL = f'wss://ws.finnhub.io?token={FINNHUB_TOKEN}'

async def subscribe(symbols):
    async with websockets.connect(WS_URL) as ws:
        # Subscribe to each symbol
        for sym in symbols:
            await ws.send(json.dumps({'type': 'subscribe', 'symbol': sym}))
        print(f'Subscribed to: {", ".join(symbols)}')

        # Listen indefinitely
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            if data.get('type') == 'trade':
                for trade in data['data']:
                    print(f"{trade['s']}: ${trade['p']} @ {trade['t']}")

# Run the coroutine
if __name__ == '__main__':
    symbols = ['AAPL', 'MSFT', 'TSLA']
    asyncio.run(subscribe(symbols))

Key points to remember:

• WebSocket connections stay open, so you only pay the initial handshake cost.
• Most providers limit the number of concurrent symbols; batch them if needed.
• Implement reconnection logic to handle network hiccups.

Best Practices for a Production‑Ready Scraper

Turning a prototype into a reliable service requires attention to error handling, logging, and scalability.

1. Rate‑Limit Management

• Read the provider’s rate‑limit documentation and enforce it programmatically (e.g., using time.sleep or token buckets).
• Cache recent responses when possible to avoid redundant calls.

2. Robust Error Handling

try:
    response = requests.get(url, timeout=5)
    response.raise_for_status()
except requests.exceptions.HTTPError as http_err:
    logger.error(f'HTTP error: {http_err}')
except requests.exceptions.Timeout:
    logger.warning('Request timed out – retrying...')
    # Implement exponential back‑off here
except Exception as err:
    logger.exception(f'Unexpected error: {err}')

3. Logging & Monitoring

Use the built‑in logging module to capture timestamps, error levels, and payload summaries. For long‑running services, ship logs to a central system like ELK or CloudWatch.

4. Data Persistence

Store fetched quotes in a time‑series database (e.g., InfluxDB, TimescaleDB) or a simple CSV/Parquet file for later analysis. Example with pandas:

df = pd.DataFrame(records)
df.to_parquet('prices.parquet', engine='pyarrow', compression='gzip')

5. Deployment Options

• Docker: Containerize the scraper for consistent environments across dev, test, and prod.
• Serverless: Use AWS Lambda (for periodic REST calls) or Azure Functions with a timer trigger.
• Dedicated VM: Keep a lightweight Linux VM running the WebSocket client 24/7.

Ensuring Legal and Ethical Compliance

Scraping financial data is subject to strict regulations. Always:

• Read the provider’s terms of service and respect robots.txt for HTML sources.
• Include attribution if required (e.g., “Data provided by Finnhub”).
• Never redistribute raw market data for commercial gain without a proper license.

Extending the Scraper: From Prices to Signals

Once you have a steady stream of live quotes, the next logical step is to generate actionable signals. Common extensions include:

• Moving‑average crossovers (e.g., SMA 20 vs. SMA 50).
• Real‑time volatility calculations using the standard deviation of the last N ticks.
• Integration with broker APIs (e.g., Alpaca, Interactive Brokers) to place orders automatically.

Below is a quick example of a 5‑minute simple moving average (SMA) calculator that updates on each new tick:

from collections import deque

class SMA:
    def __init__(self, period: int):
        self.period = period
        self.prices = deque(maxlen=period)

    def add_price(self, price: float) -> float:
        self.prices.append(price)
        return sum(self.prices) / len(self.prices)

sma_5 = SMA(5)

def on_new_tick(price):
    current_sma = sma_5.add_price(price)
    print(f'Current 5‑tick SMA: {current_sma:.2f}')

Testing and Validation

Automated tests ensure that changes to the scraper don’t break existing functionality.

• Unit tests: Mock HTTP responses with responses library and verify parsing logic
  

Looking for a way to turn endless scrolling through job boards into a streamlined, data‑driven process? A Python automated job application scraper can do just that—collecting fresh listings, filtering them to match your skill set, and even auto‑filling applications so you never miss a perfect opportunity. In this guide we’ll walk through the entire workflow, from choosing the right libraries to deploying a reliable scraper that respects legal boundaries and keeps your data tidy. Whether you’re a career‑focused developer, a recruiter building a talent pipeline, or a hobbyist eager to experiment with web automation, the steps below will give you a production‑ready blueprint you can adapt to any job site.

Why Automate Job Hunting with Python?

Job boards update every few minutes, and manually checking each site is both time‑consuming and error‑prone. Automation offers several tangible benefits:

• Speed: Gather hundreds of listings in seconds rather than hours.
• Precision: Apply custom filters (location, salary, tech stack) to surface only the most relevant roles.
• Consistency: Ensure every application follows the same format, reducing the chance of missing required fields.
• Scalability: Expand from a single board to dozens without extra manual effort.

Legal and Ethical Considerations

Before you start scraping, keep these points in mind to stay on the right side of the law and maintain good etiquette:

1. Read the Terms of Service (ToS): Many job sites explicitly forbid automated access. If the ToS disallow scraping, consider using their public API or contacting the site for permission.
2. Respect robots.txt: Although not legally binding, honoring robots.txt signals good faith and reduces the risk of being blocked.
3. Rate limiting: Insert delays (e.g., time.sleep(2)) between requests to avoid overloading the server.
4. Data privacy: Do not store personally identifiable information (PII) of other users without consent.

Choosing the Right Python Tools

The Python ecosystem offers several libraries tailored for web scraping and automation. Below is a quick comparison to help you pick the best fit for your project:

Library	Use‑Case	Pros	Cons
requests	Static HTML pages	Lightweight, easy to use	Cannot handle JavaScript
BeautifulSoup	Parsing HTML/XML	Intuitive API, great for quick extraction	Relies on another library for HTTP
Selenium	Dynamic, JavaScript‑heavy sites	Full browser automation, supports clicks & forms	Slower, requires WebDriver binaries
Playwright	Modern headless browsing	Fast, supports multiple browsers, built‑in waiting	Newer, slightly larger learning curve

Step‑by‑Step: Building a Basic Job Scraper

1. Set Up Your Environment

python -m venv job-scraper-env
source job-scraper-env/bin/activate   # Linux/macOS
.\job-scraper-env\Scripts\activate    # Windows
pip install requests beautifulsoup4 pandas
# If you need Selenium:
pip install selenium webdriver-manager

2. Fetch the Search Results Page

Start with a simple GET request to the job board’s search endpoint. Most sites let you pass query parameters for keywords, location, and pagination.

import requests
from bs4 import BeautifulSoup

BASE_URL = "https://example.com/jobs"
params = {
    "q": "python developer",
    "l": "remote",
    "page": 1
}
response = requests.get(BASE_URL, params=params, headers={"User-Agent": "Mozilla/5.0"})
soup = BeautifulSoup(response.text, "html.parser")

3. Parse Job Cards

Identify the HTML pattern that wraps each posting (often a <div class="job-card"> or similar). Extract title, company, location, and the link to the full description.

jobs = []
for card in soup.select("div.job-card"):
    title = card.select_one("h2.title").get_text(strip=True)
    company = card.select_one("span.company").get_text(strip=True)
    location = card.select_one("span.location").get_text(strip=True)
    link = card.select_one("a.apply")["href"]
    jobs.append({
        "title": title,
        "company": company,
        "location": location,
        "link": link
    })

4. Handle Pagination

Most boards use a page query parameter or a “Next” button. Loop until no new jobs are returned.

all_jobs = []
page = 1
while True:
    params["page"] = page
    resp = requests.get(BASE_URL, params=params, headers={"User-Agent": "Mozilla/5.0"})
    soup = BeautifulSoup(resp.text, "html.parser")
    new_jobs = extract_jobs(soup)   # reuse the logic from step 3
    if not new_jobs:
        break
    all_jobs.extend(new_jobs)
    page += 1
    time.sleep(2)  # polite pause

5. Filter and Store Results

Convert the list into a pandas.DataFrame for easy filtering, then export to CSV or a database.

import pandas as pd

df = pd.DataFrame(all_jobs)
# Example filter: only remote positions with "Senior" in the title
filtered = df[
    (df["location"].str.contains("Remote", case=False)) &
    (df["title"].str.contains("Senior", case=False))
]
filtered.to_csv("senior_python_remote_jobs.csv", index=False)

Automating the Application Process

Scraping listings is only half the battle; the real power comes from auto‑filling application forms. Below are two common approaches:

Using Selenium for Form Submission

• Navigate to the job’s application URL.
• Locate input fields by name or CSS selector.
• Send your resume, cover letter, and any required questionnaire answers.
• Submit and capture the confirmation message for logging.

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from webdriver_manager.chrome import ChromeDriverManager

driver = webdriver.Chrome(ChromeDriverManager().install())
driver.get(job_link)

# Example: fill email and upload resume
driver.find_element(By.NAME, "email").send_keys("you@example.com")
driver.find_element(By.NAME, "resume").send_keys("/path/to/resume.pdf")
driver.find_element(By.XPATH, "//button[text()='Submit']").click()

# Verify submission
assert "Thank you" in driver.page_source
driver.quit()

API‑Based Submissions

If the job board offers a public API (e.g., Greenhouse, Lever), you can POST a JSON payload directly, which is faster and less detectable than browser automation.

import json, requests

api_url = "https://api.example.com/applications"
payload = {
    "job_id": job_id,
    "applicant": {
        "email": "you@example.com",
        "resume_url": "https://mybucket.s3.amazonaws.com/resume.pdf"
    }
}
headers = {"Authorization": "Bearer YOUR_API_TOKEN", "Content-Type": "application/json"}
resp = requests.post(api_url, data=json.dumps(payload), headers=headers)
print(resp.status_code, resp.json())

Best Practices for a Robust Scraper

• Use rotating proxies: Prevent IP bans by rotating through a pool of residential or datacenter proxies.
• Implement exponential backoff: If you receive a 429 (Too Many Requests), wait longer before retrying.
• Log every step: Store request URLs, response codes, and errors in a log file for debugging.
• Modular code: Separate fetching, parsing, filtering, and applying into distinct functions or classes.
• Version control: Keep your scraper in a Git repository; tag releases when you add support for a new job board.

Common Pitfalls and How to Avoid Them

Even seasoned developers hit snags when building a job scraper. Here are the most frequent issues and quick fixes:

1. Dynamic content loading: If listings appear after scrolling, use Selenium’s driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") or Playwright’s page.wait_for_load_state().
2. Changing HTML structure: Sites redesign frequently. Implement CSS selectors that are tolerant to minor class name changes, or use XPath with contains() functions.
3. CAPTCHA challenges: For occasional CAPTCHAs, integrate a third‑party solving service (e.g., 2Captcha) or switch to a headless browser with human
   

Scraping dynamic web pages has become a cornerstone of modern data extraction, and Python developers often turn to Pyppeteer for a seamless, headless-browser experience. In this guide, we’ll explore how to master Python dynamic content scraping with Pyppeteer, from installation to advanced techniques that keep your bots fast, reliable, and SEO‑friendly.

Why Choose Pyppeteer for Dynamic Content Scraping?

Traditional libraries like requests and BeautifulSoup excel at static HTML, but they stumble when a page relies on JavaScript to render its data. Pyppeteer, the Python port of Google’s Puppeteer, controls a Chromium instance programmatically, allowing you to:

• Render JavaScript‑heavy sites exactly as a real browser would.
• Interact with page elements (click, scroll, type) to trigger lazy‑loaded content.
• Capture screenshots or PDFs for visual verification.
• Bypass basic anti‑scraping measures that block non‑browser traffic.

Getting Started: Installing Pyppeteer

Pyppeteer can be installed via pip. It automatically downloads a compatible Chromium binary, so you’re ready to scrape within minutes.

pip install pyppeteer
# Optional: install a specific Chromium version
# pip install pyppeteer==0.2.5

On Linux servers, you may need additional system dependencies for Chromium to run headlessly:

sudo apt-get update
sudo apt-get install -y \
    libnss3 libatk-bridge2.0-0 libx11-xcb1 \
    libxcomposite1 libxcursor1 libxdamage1 \
    libxrandr2 libgbm1 libasound2

Basic Pyppeteer Workflow

1. Launching a Browser Instance

Start with a minimal script that opens a headless Chromium window and navigates to a URL.

import asyncio
from pyppeteer import launch

async def main():
    browser = await launch(headless=True)  # Set to False for debugging
    page = await browser.newPage()
    await page.goto('https://example.com')
    content = await page.content()
    print(content[:500])  # Print first 500 characters
    await browser.close()

asyncio.get_event_loop().run_until_complete(main())

2. Waiting for Dynamic Elements

Dynamic pages often load content after the initial HTML response. Use page.waitForSelector() or page.waitForFunction() to pause execution until the desired element appears.

await page.waitForSelector('#price', {'visible': True})
price = await page.evaluate('''() => {
    return document.querySelector('#price').innerText;
}''')
print('Current price:', price)

3. Extracting Structured Data

Once the page is fully rendered, you can pull data using page.evaluate() (executing JavaScript) or by parsing the HTML with BeautifulSoup.

from bs4 import BeautifulSoup

html = await page.content()
soup = BeautifulSoup(html, 'html.parser')
titles = [t.text.strip() for t in soup.select('.article-title')]
print(titles)

Handling Complex Dynamic Scenarios

Infinite Scrolling and Lazy Loading

Many modern sites load additional items as you scroll. The following pattern scrolls to the bottom of the page repeatedly until no new content appears.

async def auto_scroll(page):
    await page.evaluate('''
        async () => {
            await new Promise(resolve => {
                let totalHeight = 0;
                const distance = 100;
                const timer = setInterval(() => {
                    window.scrollBy(0, distance);
                    totalHeight += distance;
                    if (totalHeight >= document.body.scrollHeight) {
                        clearInterval(timer);
                        resolve();
                    }
                }, 200);
            });
        }
    ''')

Combine auto_scroll with a loop that checks for new items to ensure you capture the entire dataset.

Dealing with AJAX Calls

Instead of waiting for the UI to update, you can intercept network requests and extract JSON responses directly.

responses = []

def intercept_response(response):
    if 'api/products' in response.url:
        responses.append(response)

await page.setRequestInterception(True)
page.on('response', intercept_response)

await page.goto('https://shop.example.com')
await page.waitForTimeout(5000)  # Give AJAX time to finish

for r in responses:
    data = await r.json()
    print(data['items'])

Bypassing Simple Bot Protections

Some sites check for headless browsers via the navigator.webdriver flag. Pyppeteer lets you spoof this flag:

await page.evaluateOnNewDocument('''() => {
    Object.defineProperty(navigator, 'webdriver', {
        get: () => false,
    });
}''')

Combine this with a realistic User-Agent and random delays to further reduce detection risk.

Advanced Techniques for Production‑Ready Scrapers

Parallel Page Processing

Running multiple pages concurrently maximizes throughput. Use asyncio.gather() to launch several scraping tasks in parallel.

async def scrape_product(url):
    page = await browser.newPage()
    await page.goto(url)
    await page.waitForSelector('.price')
    price = await page.evaluate('''() => document.querySelector('.price').innerText''')
    await page.close()
    return {'url': url, 'price': price}

urls = ['https://site.com/p1', 'https://site.com/p2', 'https://site.com/p3']
results = await asyncio.gather(*[scrape_product(u) for u in urls])
print(results)

Persisting Sessions with Cookies and Local Storage

If your target site requires login, store the authentication cookies and reuse them across runs.

# After manual login
cookies = await page.cookies()
import json, os
with open('cookies.json', 'w') as f:
    json.dump(cookies, f)

# Reuse later
with open('cookies.json') as f:
    cookies = json.load(f)
await page.setCookie(*cookies)
await page.goto('https://protected.example.com')

Graceful Error Handling and Retries

Network hiccups or anti‑scraping blocks can cause failures. Wrap critical sections in try/except blocks and implement exponential back‑off.

import random, time

async def safe_goto(page, url, retries=3):
    for attempt in range(retries):
        try:
            await page.goto(url, {'waitUntil': 'networkidle2'})
            return True
        except Exception as e:
            wait = 2 ** attempt + random.random()
            print(f'Attempt {attempt+1} failed: {e}. Retrying in {wait:.2f}s')
            await asyncio.sleep(wait)
    return False

Best Practices for SEO‑Friendly Scraping

• Respect robots.txt: Before crawling, check the site’s robots.txt and honor disallowed paths.
• Throttle Requests: Use random delays (1‑5 seconds) between page loads to mimic human behavior.
• Identify Your Bot: Include a meaningful User-Agent and, if possible, an email address for site owners to contact.
• Cache Results: Store scraped data locally or in a database to avoid unnecessary repeated requests.
• Monitor Performance: Log response times, error rates, and resource usage to quickly spot bottlenecks.

Common Pitfalls and How to Avoid Them

Memory Leaks in Long‑Running Sessions

Each opened page consumes memory. Always close pages with await page.close() and, for very long jobs, periodically restart the browser instance.

Unexpected Pop‑ups or Modals

Pop‑ups can block selectors. Detect them early and dismiss using page.click() or by evaluating JavaScript to remove the element.

Changing DOM Structures

Websites frequently update their markup. Use robust selectors (e.g., data attributes) and consider fallback strategies like searching by text content.

Putting It All Together: A Complete Sample Scraper

The following script demonstrates a production‑ready workflow that logs in, scrolls through an infinite‑scroll product list, extracts JSON data from intercepted AJAX calls, and saves the results to a CSV file.

import asyncio, csv, json

from pyppeteer import launch

async def login(page, username, password):

    await page.goto('https://shop.example.com/login')

    await page.type('#email', username)

    await page.type('#pwd', password)

    await page.click('#loginBtn')

    await page.waitForNavigation({'waitUntil': 'networkidle2'})

    # Save cookies for later reuse

    cookies = await page.cookies()

    with open('session_cookies.json', 'w') as f:

        json.dump(cookies, f)
async def scrape_products(page):

    products = []

    ajax_responses = []
    async def capture_response(response):

        if 'api/v1/products' in response.url:

            ajax_responses.append(response)
    page.on('response', capture_response)

    await page.goto('https://shop.example.com/catalog')

    await page.waitForSelector('.product-card')

    await auto_scroll(page)  # Defined earlier
    for resp in ajax_responses:

        data = await resp.json()

        for item in data['items']:

            products.append({

                'id': item['id'],

                '