In modern data-driven environments, collecting data consistently from web sources without paying for heavy server infrastructure is a major engineering advantage. Serverless architectures coupled with CI/CD platforms like GitHub Actions provide an elegant, cost-effective, and highly reliable way to run scheduled data scraping pipelines.
In this comprehensive tutorial, we will build an enterprise-grade, serverless web scraper and webhook notification pipeline using Python 3, BeautifulSoup4, Requests, and GitHub Actions.
Architecture Overview
-
Scraper Core (
scraper.py): Fetches target web content using robust HTTP session retries, custom User-Agents, and rate-limiting. - Parser Module: Extracts structured data efficiently with BeautifulSoup, handling missing DOM nodes gracefully.
- Webhook Notifier: Posts newly discovered or updated records to Discord/Slack/Custom Webhook endpoints via JSON payload.
-
CI/CD Serverless Runner (
.github/workflows/scraper.yml): Executes on a scheduled cron trigger (e.g., hourly/daily) or manual dispatch, persisting log output and caching HTTP responses if necessary.
Core Python Scraper Engine (scraper.py)
Below is the complete, self-contained Python implementation:
import os
import sys
import time
import json
import logging
import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
# Setup Structured Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("ServerlessScraper")
def create_resilient_session(retries=3, backoff_factor=1.5, status_forcelist=(500, 502, 503, 504)):
"""Configures a requests Session with automatic exponential backoff retries."""
session = requests.Session()
retry_strategy = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
raise_on_status=False
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
session.headers.update({
"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-Language": "en-US,en;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
})
return session
def parse_target_data(html_content):
"""Parses target HTML with BeautifulSoup and extracts structured items."""
soup = BeautifulSoup(html_content, "html.parser")
extracted_items = []
articles = soup.find_all(["article", "div"], class_=lambda c: c and any(term in str(c) for term in ["card", "item", "post"]))
for idx, item in enumerate(articles[:10]):
title_elem = item.find(["h1", "h2", "h3", "a"])
link_elem = item.find("a", href=True)
if title_elem:
title = title_elem.get_text(strip=True)
link = link_elem["href"] if link_elem else ""
if link and not link.startswith("http"):
link = f"https://news.ycombinator.com/{link}"
extracted_items.append({
"id": idx + 1,
"title": title,
"url": link
})
return extracted_items
def send_webhook_notification(webhook_url, data):
"""Sends JSON payload to configured Webhook (Slack/Discord/Custom)."""
if not webhook_url:
logger.warning("No WEBHOOK_URL configured. Skipping notification step.")
return False
payload = {
"content": f"🚀 **Serverless Scraper Pipeline Report**\nSuccessfully extracted {len(data)} items.",
"embeds": [
{
"title": item["title"][:256],
"url": item["url"],
"color": 3447003
} for item in data[:5]
]
}
try:
response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()
logger.info("Webhook notification delivered successfully.")
return True
except Exception as e:
logger.error(f"Failed to send webhook notification: {e}")
return False
def main():
target_url = os.getenv("TARGET_URL", "https://news.ycombinator.com/")
webhook_url = os.getenv("WEBHOOK_URL", "")
logger.info(f"Starting resilient extraction run for: {target_url}")
session = create_resilient_session()
try:
response = session.get(target_url, timeout=15)
response.raise_for_status()
logger.info(f"Received HTTP {response.status_code}. Parsing DOM structure...")
parsed_data = parse_target_data(response.text)
logger.info(f"Successfully extracted {len(parsed_data)} items.")
print(json.dumps(parsed_data, indent=2))
if webhook_url:
send_webhook_notification(webhook_url, parsed_data)
except requests.RequestException as err:
logger.critical(f"Network error during execution: {err}")
sys.exit(1)
if __name__ == "__main__":
main()
GitHub Actions Automated Workflow (.github/workflows/scraper.yml)
name: Resilient Serverless Web Scraper
on:
schedule:
- cron: '0 */6 * * *'
workflow_dispatch:
jobs:
scrape-and-notify:
runs-on: ubuntu-latest
steps:
- name: Checkout Code Repository
uses: actions/checkout@v4
- name: Set up Python 3.11 Environment
uses: actions/setup-python@v5
with:
python-version: '3.11'
cache: 'pip'
- name: Install Python Dependencies
run: |
python -m pip install --upgrade pip
pip install requests beautifulsoup4 urllib3
- name: Execute Serverless Scraper Pipeline
env:
TARGET_URL: 'https://news.ycombinator.com/'
WEBHOOK_URL: ${{ secrets.WEBHOOK_URL }}
run: |
python scraper.py
Best Practices & Resilience Strategies
- Exponential Backoff Retries: Handles temporary network degradation cleanly.
- User-Agent & Header Rotation: Simulates real browser handshake headers.
-
Secret Management: Store webhook endpoints in
secrets.WEBHOOK_URL. - Zero Cost: Completely free with GitHub Actions runners.
Top comments (0)