Building Production Web Scraping Pipelines With Python

· 6 min read ·
Building Production Web Scraping Pipelines With Python

What You’ll Need

  • Hetzner VPS or Contabo VPS for hosting production worker nodes
  • DigitalOcean as an alternative cloud hosting platform
  • n8n Cloud or self-hosted n8n for orchestration and alerts
  • Python 3.11 or higher installed on your build machine
  • Docker Engine and Docker Compose installed on your deployment server

Table of Contents

Architecture of a Resilient Scraping Pipeline

Writing a script that fetches HTML with a basic HTTP library works fine for small local projects. Once you scale to hundreds of thousands of requests per day, simple scripts break quickly. Web applications block IP addresses, alter DOM structures without notice, leak browser memory, and throw unhandled timeout exceptions.

To run web scrapers reliably in production, you must treat data collection like any other critical backend subsystem. I split production pipelines into four distinct layers:

  1. Extraction Layer: Playwright manages headless Chromium instances to execute JavaScript, render dynamic content, and simulate real user interactions.
  2. Validation Layer: Pydantic enforces strict data schemas on extracted fields before anything enters the database.
  3. Queue and Task Layer: Celery and Redis manage distributed request processing, handle exponential backoff retries, and rate-limit outgoing jobs.
  4. Persistence Layer: PostgreSQL stores fully validated records with unique constraints to prevent duplicate entries.

When running multiple headless Chromium instances inside worker nodes, hardware resources deplete rapidly. Chrome processes consume substantial RAM, which can trigger Out-Of-Memory (OOM) kernel kills on budget nodes. You can mitigate this instability by configuring swap space on Ubuntu Linux VPS to handle unexpected RAM spikes without crashing your worker containers.

       +-------------------------------------------------------+
       |                  Producer / Scheduler                 |
       +-------------------------------------------------------+
                                   |
                                   v
       +-------------------------------------------------------+
       |                  Redis Message Broker                 |
       +-------------------------------------------------------+
                                   |
                                   v
       +-------------------------------------------------------+
       |                 Celery Worker Pool                    |
       |  +-------------------------------------------------+  |
       |  | Playwright Engine (Headless Browser Render)     |  |
       |  +-------------------------------------------------+  |
       |  | Pydantic Validator (Type Enforcement)           |  |
       |  +-------------------------------------------------+  |
       +-------------------------------------------------------+
                                   |
                                   v
       +-------------------------------------------------------+
       |                 PostgreSQL Database                   |
       +-------------------------------------------------------+

Building the Async Extractor with Playwright and Pydantic

The extraction module opens a headless browser session, injects custom headers, handles page load events, and transforms unorganized HTML elements into verified Python models.

First, create a virtual environment and install the required dependencies:

pip install playwright pydantic psycopg2-binary celery redis
playwright install chromium

Create a file named extractor.py containing the core Playwright logic and Pydantic schemas. I use Pydantic to ensure that if a website changes its HTML structure and returns null values, the parser fails early rather than corrupting the database.

import asyncio
import logging
from typing import Optional
from playwright.async_api import async_playwright, Browser, Page
from pydantic import BaseModel, Field, HttpUrl, ValidationError

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

class ScrapedProduct(BaseModel):
    title: str = Field(min_length=1)
    price: float = Field(gt=0.0)
    in_stock: bool
    url: HttpUrl
    sku: str = Field(min_length=3)

class ScraperEngine:
    def __init__(self, proxy_server: Optional[str] = None):
        self.proxy_server = proxy_server

    async def extract_product(self, target_url: str) -> Optional[ScrapedProduct]:
        async with async_playwright() as p:
            launch_options = {
                "headless": True,
                "args": ["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"]
            }
            if self.proxy_server:
                launch_options["proxy"] = {"server": self.proxy_server}

            browser: Browser = await p.chromium.launch(**launch_options)
            context = await browser.new_context(
                user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
                viewport={"width": 1920, "height": 1080}
            )
            page: Page = await context.new_page()

            try:
                logging.info(f"Navigating to: {target_url}")
                await page.goto(target_url, wait_until="networkidle", timeout=30000)
                
                raw_title = await page.locator("h1.product-title").inner_text()
                raw_price = await page.locator("span.price-value").inner_text()
                stock_text = await page.locator("div.stock-status").inner_text()
                raw_sku = await page.locator("span.sku-code").inner_text()

                clean_price = float(raw_price.replace("$", "").replace(",", "").strip())
                is_in_stock = "in stock" in stock_text.lower()

                product = ScrapedProduct(
                    title=raw_title.strip(),
                    price=clean_price,
                    in_stock=is_in_stock,
                    url=target_url,
                    sku=raw_sku.strip()
                )
                
                await browser.close()
                return product

            except ValidationError as ve:
                logging.error(f"Validation error parsing {target_url}: {ve}")
                await browser.close()
                raise ve
            except Exception as e:
                logging.error(f"Extraction error on {target_url}: {e}")
                await browser.close()
                raise e

if __name__ == "__main__":
    test_url = "https://quotes.toscrape.com/"
    engine = ScraperEngine()
    print("Extractor initialized for execution.")

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Scaling with Celery, Redis, and Proxy Rotation

To scale extraction across thousands of pages without running out of resources, wrap the browser engine in asynchronous worker tasks using Celery. Redis acts as our message queue broker, storing incoming job URLs and dispatching them to available worker threads.

Deploying worker pools on isolated hardware like a Hetzner VPS guarantees dedicated CPU resources for heavy browser parsing tasks.

Create a file named tasks.py to handle queue processing, exponential retry logic, and PostgreSQL persistence:

import os
import asyncio
import psycopg2
from celery import Celery
from extractor import ScraperEngine, ScrapedProduct

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/scraper_db")

celery_app = Celery("scraper_tasks", broker=REDIS_URL, backend=REDIS_URL)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,
    worker_prefetch_multiplier=1
)

def get_db_connection():
    return psycopg2.connect(DATABASE_URL)

def init_db():
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS scraped_products (
            id SERIAL PRIMARY KEY,
            sku VARCHAR(100) UNIQUE NOT NULL,
            title TEXT NOT NULL,
            price NUMERIC(10, 2) NOT NULL,
            in_stock BOOLEAN NOT NULL,
            url TEXT NOT NULL,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
    """)
    conn.commit()
    cursor.close()
    conn.close()

@celery_app.task(bind=True, max_retries=3, default_retry_delay=15)
def scrape_url_task(self, target_url: str, proxy: str = None):
    try:
        engine = ScraperEngine(proxy_server=proxy)
        product_data: ScrapedProduct = asyncio.run(engine.extract_product(target_url))
        
        if product_data:
            conn = get_db_connection()
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO scraped_products (sku, title, price, in_stock, url, updated_at)
                VALUES (%s, %s, %s, %s, %s, NOW())
                ON CONFLICT (sku) DO UPDATE SET
                    title = EXCLUDED.title,
                    price = EXCLUDED.price,
                    in_stock = EXCLUDED.in_stock,
                    url = EXCLUDED.url,
                    updated_at = NOW();
            """, (
                product_data.sku,
                product_data.title,
                product_data.price,
                product_data.in_stock,
                str(product_data.url)
            ))
            conn.commit()
            cursor.close()
            conn.close()
            return {"status": "success", "sku": product_data.sku}

    except Exception as exc:
        logging.warning(f"Task failed for {target_url}. Retrying... Details: {exc}")
        raise self.retry(exc=exc)

If jobs repeatedly fail after maximum retries, configured alerting mechanisms can send instant notifications to monitoring groups. You can integrate automated alerting channels by deploying self hosted Telegram bots with Docker alongside your Celery cluster.

Deploying and Orchestrating the Pipeline

To deploy the worker system reliably, containerize the setup using Docker and Docker Compose. This strategy standardizes browser dependencies across development and production environments.

Create a Dockerfile in your project root:

FROM python:3.11-slim

ENV PYTHONUNBUFFERED=1 \
    DEBIAN_FRONTEND=noninteractive

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    libpq-dev \
    curl \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install-deps chromium
RUN playwright install chromium

COPY . .

CMD ["celery", "-A", "tasks.celery_app", "worker", "--loglevel=info", "--concurrency=2"]

Next, create a docker-compose.yml configuration file to run PostgreSQL, Redis, and your Celery scraping workers simultaneously:

version: '3.8'

services:
  postgres:
    image: postgres:15-alpine
    container_name: scraper_postgres
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgrespassword
      POSTGRES_DB: scraper_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    container_name: scraper_redis
    ports:
      - "6379:6379"

  worker:
    build: .
    container_name: scraper_worker
    command: celery -A tasks.celery_app worker --loglevel=info --concurrency=2
    depends_on:
      - postgres
      - redis
    environment:
      - REDIS_URL=redis://redis:6379/0
      - DATABASE_URL=postgresql://postgres:postgrespassword@postgres:5432/scraper_db
    restart: always

volumes:
  pgdata:

If you manage broad automation infrastructure, hosting these containers on cloud VPS hosting like Contabo VPS provides cheap multi-core CPU capacity for high-concurrency Chromium processing. For structured orchestration across your microservices, read our guide on deploying open source workflow systems on Hetzner to schedule worker execution automatically.

Launch the full platform using Docker Compose:

docker compose up -d --build

You can dispatch incoming jobs from any Python terminal or scheduler script:

from tasks import scrape_url_task

job = scrape_url_task.delay(
    target_url="https://example.com/product/123",
    proxy="http://user:password@proxy.example.com:8080"
)
print(f"Task dispatched with ID: {job.id}")

Getting Started

To launch your production scraping cluster, set up server infrastructure using cloud hosts like a Hetzner VPS or Contabo VPS. If you prefer managed node infrastructure, evaluate providers like DigitalOcean. Once server instances are active, connect incoming extracted feeds into downstream workflow layers using n8n Cloud to distribute alerts and generate automated business intelligence reports.

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr, mention SYS3-HUGO for priority. Or DM at chasebot.online.

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online