PostgreSQL-Backed Scraper Pipeline
Build a two-pass Playwright scraper coordinated by a PostgreSQL task queue.
Introduction
30 Second Summary
When companies need to gather data from thousands of web pages, they don't just loop through URLs and hope for the best. They build pipelines with job queues that track progress, prevent duplicate work, and recover gracefully from failures.
In this project, you will build a containerized book scraping pipeline that uses PostgreSQL as both its data store and its job queue. You'll scrape books.toscrape.com in two passes, preserve raw HTML for future enrichment, and coordinate everything through atomic task claims.
What You'll Build
A Docker Compose stack with PostgreSQL 16 that powers a two-pass Playwright scraper, using SKIP LOCKED queries to coordinate task distribution without any external message broker.
By the end of this project, you'll have:
- A PostgreSQL task queue that atomically claims jobs using FOR UPDATE SKIP LOCKED, preventing duplicate work across concurrent workers.
- A two-pass Playwright scraper that discovers books from listing pages, then extracts full metadata and raw HTML from detail pages.
- A Dockerized pipeline with built-in rate limiting and priority-based job scheduling.
- Secret Mission: weighted round-robin scheduling that balances scrape effort across book categories with interleaved deep-scrape draining.
Are there any prerequisites?
This project assumes familiarity with Python async/await, basic Docker usage (running containers, reading a Compose file), and SQL fundamentals (SELECT, INSERT, UPDATE). You don't need prior experience with Playwright or PostgreSQL-based queues.
Set Up the Project and Docker Stack
Your Python environment is ready with the scraping and database libraries installed. The next piece is the database itself. PostgreSQL will serve as both the data store for scraped books and the job queue that coordinates scraping work.
Docker Compose lets you define the entire database stack in a single YAML file. One command spins up a configured PostgreSQL instance with persistent storage, healthchecks, and predictable credentials.
In this step, get ready to:
- Create a Docker Compose file that defines the PostgreSQL 16 service.
- Start the database container and verify it's healthy.
- Build a Python connection module and confirm connectivity.
Create the Docker Compose file
Docker Compose reads a docker-compose.yml file that declares which containers to run and how to configure them. You need a single PostgreSQL 16 container with a healthcheck so your pipeline knows the database is ready before trying to connect.
- Create a docker-compose.yml file in the root of your book-scraper folder with this content:
services:
db:
image: postgres:16
environment:
POSTGRES_USER: scraper
POSTGRES_PASSWORD: scraper_pass
POSTGRES_DB: bookpipeline
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U scraper -d bookpipeline"]
interval: 5s
timeout: 5s
retries: 5
volumes:
pgdata:What does this file do?
- image: postgres:16 pulls the official PostgreSQL 16 image from Docker Hub.
- environment sets the database name, user, and password that PostgreSQL creates on first boot.
- ports exposes port 5432 on your host so Python can reach the database at localhost:5432.
- volumes maps a named volume (pgdata) to PostgreSQL's data directory so your data survives container restarts.
- healthcheck runs pg_isready every 5 seconds to confirm the database is accepting connections.
- Save docker-compose.yml.
- Confirm the file exists by running:
ls docker-compose.ymlYou should see docker-compose.yml in the output.
File not showing up?
- Make sure you saved the file in the root of your book-scraper folder, not inside a subdirectory.
- Check that the filename is exactly docker-compose.yml with no extra extension like .txt.
Why can't Docker Compose find my file?
✔️ Awesome, I've got everything!
Great. Make sure you've saved docker-compose.yml before moving on.
ⓧ I'd like to double check the full code
services:
db:
image: postgres:16
environment:
POSTGRES_USER: scraper
POSTGRES_PASSWORD: scraper_pass
POSTGRES_DB: bookpipeline
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U scraper -d bookpipeline"]
interval: 5s
timeout: 5s
retries: 5
volumes:
pgdata:Start the PostgreSQL container
With Docker already running, you can bring up the stack in detached mode. The --wait flag blocks until the healthcheck passes, so you know the database is ready.
- Start the PostgreSQL container by running:
docker compose up -d --waitAfter a few seconds, you should see a message confirming the container is healthy.
- Verify the container status by running:
docker compose psYou should see the db service listed with a status that includes (healthy).
Container not healthy?
- If the status shows (health: starting), wait 10-15 seconds and run docker compose ps again. The healthcheck retries up to 5 times.
- If the container exited, check the logs by running docker compose logs db and look for port conflicts or permission errors.
- Make sure no other service is already using port 5432 on your machine.
Help me debug my Docker Compose container that won't start.
Build the connection module and verify connectivity
Your pipeline needs a reusable function to connect to PostgreSQL. You will create a small Python package structure and a connection module that the rest of the project imports.
- Create the package directories and init files by running:
mkdir -p src/db
touch src/__init__.py src/db/__init__.py- Confirm the structure exists by running:
ls src src/dbYou should see __init__.py listed inside both src/ and src/db/.
- Create the connection module at src/db/connection.py with this content:
import psycopg2
from psycopg2.extras import RealDictCursor
def get_connection():
return psycopg2.connect(
host="localhost",
port=5432,
dbname="bookpipeline",
user="scraper",
password="scraper_pass",
cursor_factory=RealDictCursor,
)What does this code do?
- get_connection() returns a new connection to the bookpipeline database using the credentials you set in docker-compose.yml.
- RealDictCursor makes query results return as Python dictionaries instead of tuples. This lets you access columns by name (row["title"]) instead of index.
- Every module in the project will call get_connection() instead of duplicating connection logic.
- Save src/db/connection.py.
- Test the connection by running:
uv run python -c "from src.db.connection import get_connection; conn = get_connection(); print('Connected to bookpipeline!'); conn.close()"You should see Connected to bookpipeline! printed in your terminal.
Seeing a connection refused error?
- Make sure the Docker container is still running with docker compose ps. If it stopped, restart it with docker compose up -d --wait.
- Verify the port mapping shows 5432:5432 in the docker compose ps output.
- Check that your connection credentials in connection.py match the environment variables in docker-compose.yml exactly.
Help me fix my psycopg2 connection error.
✔️ Awesome, I've got everything!
Great. Make sure you've saved src/db/connection.py before moving on.
ⓧ I'd like to double check the full code
import psycopg2
from psycopg2.extras import RealDictCursor
def get_connection():
return psycopg2.connect(
host="localhost",
port=5432,
dbname="bookpipeline",
user="scraper",
password="scraper_pass",
cursor_factory=RealDictCursor,
)Your database is running and Python can reach it. Next up, you will design the PostgreSQL schema that gives the pipeline its task queue and book storage tables.
Design the PostgreSQL Schema
Your PostgreSQL database is running, but it's completely empty. Before the pipeline can queue jobs or store scraped books, it needs a structure that separates concerns: one table for orchestrating work, another for holding the data.
In this step, you'll define a schema with two tables. The tasks table acts as the job queue, tracking what needs to be scraped, its priority, and its current status. The books table stores the final scraped data alongside raw HTML for future enrichment.
In this step, get ready to:
- Define a tasks table for the job queue with priority, status, and rate-limiting fields.
- Define a books table that preserves raw HTML alongside structured metadata.
- Add performance indexes for efficient SKIP LOCKED job claiming.
Define the task queue and books tables
The schema needs to handle two distinct responsibilities. The task queue table tracks every scraping job through its lifecycle (pending, in progress, completed, or failed). The books table stores the final output of each scrape, including a full copy of the raw HTML.
- Create a new file called src/db/schema.py and paste the following schema definition:
from src.db.connection import get_connection
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS tasks (
id SERIAL PRIMARY KEY,
task_type TEXT NOT NULL,
payload JSONB NOT NULL,
priority INTEGER DEFAULT 0,
status TEXT DEFAULT 'pending',
domain TEXT DEFAULT 'books.toscrape.com',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
started_at TIMESTAMP,
completed_at TIMESTAMP,
error TEXT
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_payload_url
ON tasks ((payload->>'url'))
WHERE status IN ('pending', 'in_progress');
CREATE INDEX IF NOT EXISTS idx_tasks_queue
ON tasks (status, priority DESC, created_at ASC);
CREATE TABLE IF NOT EXISTS books (
id SERIAL PRIMARY KEY,
url TEXT UNIQUE NOT NULL,
title TEXT,
price TEXT,
rating TEXT,
category TEXT,
availability TEXT,
upc TEXT,
description TEXT,
raw_html TEXT,
scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""What does this schema do?
The tasks table is the job queue. Each row represents one scraping job:
- payload is JSONB, so each task can carry flexible data (like a URL and metadata) without needing extra columns.
- priority lets the queue serve high-priority jobs first (higher number = higher priority).
- domain tracks which site a job targets, enabling per-domain rate limiting.
- status moves through pending to in_progress to completed (or failed).
The idx_tasks_payload_url unique index prevents the same URL from being enqueued twice while it's still pending or in progress. The idx_tasks_queue index lets SELECT FOR UPDATE SKIP LOCKED queries find the next highest-priority job without scanning the whole table.
The books table stores structured fields (title, price, category) alongside the full raw_html. Keeping raw HTML means you can re-extract data later without re-scraping the site.
Why preserve raw HTML?
Scraping is expensive and fragile. Sites change, rate-limit you, or go down. By storing the full page content on the first pass, future enrichment stages (like an LLM extracting sentiment or summaries) can work from the saved HTML without hitting the live site again.
- You won't see a change yet. The next section adds the function that executes this SQL against PostgreSQL.
Apply the schema to PostgreSQL
The SQL string defines the schema, but nothing has been sent to the database yet. You need a function that opens a connection, executes the SQL, and commits the transaction.
- Add the following code below the SCHEMA_SQL string in src/db/schema.py:
def create_schema():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(SCHEMA_SQL)
conn.commit()
print("Schema created successfully.")
finally:
conn.close()
if __name__ == "__main__":
create_schema()What does this function do?
- create_schema() opens a connection using the get_connection() helper you built in Step 1, executes the entire SQL string, and commits.
- The try/finally block ensures the connection closes even if the SQL fails.
- The if __name__ == "__main__" guard lets you run this file directly to create the schema.
- Save src/db/schema.py.
- Run the schema creation script by running this command:
uv run python -m src.db.schema- You should see Schema created successfully. printed in the terminal.
Seeing a connection error?
- Make sure your Docker Compose stack is still running. Run docker compose ps and confirm the db service shows as healthy.
- Check that port 5432 isn't in use by another PostgreSQL instance on your machine.
- Verify the credentials in src/db/connection.py match what you set in docker-compose.yml (user: scraper, password: scraper_pass, database: bookpipeline).
Help me debug my PostgreSQL connection error when running the schema script.
- Verify the tables were created by running this command:
docker compose exec db psql -U scraper -d bookpipeline -c "\dt"- You should see both tasks and books listed in the output.
What is SELECT FOR UPDATE SKIP LOCKED?
This is the core primitive that makes PostgreSQL work as a task queue. When a worker claims a job, it locks that row. If a second worker tries to claim at the same time, SKIP LOCKED tells it to skip the locked row and grab the next available one instead of waiting.
This means multiple workers never duplicate the same job, and nobody blocks waiting for a lock to release. The idx_tasks_queue index you just created makes these queries fast by pre-sorting pending tasks by priority.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file.
ⓧ I'd like to double check the full code
from src.db.connection import get_connection
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS tasks (
id SERIAL PRIMARY KEY,
task_type TEXT NOT NULL,
payload JSONB NOT NULL,
priority INTEGER DEFAULT 0,
status TEXT DEFAULT 'pending',
domain TEXT DEFAULT 'books.toscrape.com',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
started_at TIMESTAMP,
completed_at TIMESTAMP,
error TEXT
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_payload_url
ON tasks ((payload->>'url'))
WHERE status IN ('pending', 'in_progress');
CREATE INDEX IF NOT EXISTS idx_tasks_queue
ON tasks (status, priority DESC, created_at ASC);
CREATE TABLE IF NOT EXISTS books (
id SERIAL PRIMARY KEY,
url TEXT UNIQUE NOT NULL,
title TEXT,
price TEXT,
rating TEXT,
category TEXT,
availability TEXT,
upc TEXT,
description TEXT,
raw_html TEXT,
scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
def create_schema():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(SCHEMA_SQL)
conn.commit()
print("Schema created successfully.")
finally:
conn.close()
if __name__ == "__main__":
create_schema()Your database now has the structure to hold both task orchestration data and scraped book content. Next up, you'll build the task queue engine that uses SKIP LOCKED to atomically claim jobs from this table.
Build the Task Queue Engine
Your schema defines where tasks live, but you need code that manages the lifecycle of each task. Inserting, claiming, completing, and failing jobs all require careful transaction handling to prevent duplicate work.
In this step, you will build a task queue engine in Python that talks directly to your PostgreSQL tasks table. The key technique is FOR UPDATE SKIP LOCKED, a PostgreSQL feature that lets multiple workers claim jobs without blocking each other or processing the same job twice.
In this step, get ready to:
- Implement an enqueue function that inserts jobs with deduplication.
- Implement an atomic claim function using SKIP LOCKED.
- Add status transition helpers and a rate limiter.
Implement the enqueue function with deduplication
The first job of the queue engine is accepting new work. When the scraper discovers a URL it wants to visit, it needs to add that URL as a task. If the same URL is already pending, the queue should silently ignore the duplicate.
- Create the src/queue/ package by running these commands:
mkdir src/queue
touch src/queue/__init__.py- Confirm the files exist by running:
ls src/queueYou should see __init__.py in the output.
- Create src/queue/task_queue.py and add the following code:
import json
import time
from src.db.connection import get_connection
RATE_LIMIT_SECONDS = 2.0
_last_request_time = 0.0
def enqueue_task(task_type, payload, priority=0, domain="books.toscrape.com"):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO tasks (task_type, payload, priority, domain)
VALUES (%s, %s, %s, %s)
ON CONFLICT ((payload->>'url'))
WHERE status IN ('pending', 'in_progress')
DO NOTHING
RETURNING id
""",
(task_type, json.dumps(payload), priority, domain),
)
result = cur.fetchone()
conn.commit()
return result["id"] if result else None
finally:
conn.close()What does this code do?
The top of the file imports json for serializing payloads and time for the rate limiter you will add later. RATE_LIMIT_SECONDS and _last_request_time are module-level values used by the rate limiting function.
The enqueue_task() function inserts a new row into the tasks table. The ON CONFLICT DO NOTHING clause leverages your partial unique index. If a task with the same URL is already pending or in progress, the insert silently does nothing and returns None instead of raising an error.
- Save src/queue/task_queue.py.
- Verify the module loads correctly by running:
uv run python -c "from src.queue.task_queue import enqueue_task; print('enqueue_task loaded')"You should see enqueue_task loaded printed to your terminal.
Seeing a ModuleNotFoundError?
Make sure the __init__.py file exists in both src/ and src/queue/. Python needs these files to recognize the directories as packages.
Also check that you are running the command from the project root (the folder containing pyproject.toml).
Help me debug this import error
Implement atomic task claiming with SKIP LOCKED
Claiming a task means a worker picks up the next available job and marks it as in progress. The critical requirement is atomicity. If two workers try to claim at the same time, each must get a different task. FOR UPDATE SKIP LOCKED achieves this by skipping any row already locked by another transaction.
- In src/queue/task_queue.py, add the claim_task() function below enqueue_task():
def claim_task(task_type=None):
conn = get_connection()
try:
with conn.cursor() as cur:
if task_type:
cur.execute(
"""
WITH next_task AS (
SELECT id
FROM tasks
WHERE status = 'pending'
AND task_type = %s
ORDER BY priority DESC, created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE tasks
SET status = 'in_progress', started_at = CURRENT_TIMESTAMP
FROM next_task
WHERE tasks.id = next_task.id
RETURNING tasks.*
""",
(task_type,),
)
else:
cur.execute(
"""
WITH next_task AS (
SELECT id
FROM tasks
WHERE status = 'pending'
ORDER BY priority DESC, created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE tasks
SET status = 'in_progress', started_at = CURRENT_TIMESTAMP
FROM next_task
WHERE tasks.id = next_task.id
RETURNING tasks.*
"""
)
result = cur.fetchone()
conn.commit()
return result
finally:
conn.close()How does the atomic claim work?
The function uses a Common Table Expression (CTE) named next_task to SELECT the highest-priority pending task. The FOR UPDATE SKIP LOCKED clause locks that row and skips it if another transaction already holds a lock on it.
The outer UPDATE immediately sets the status to in_progress and records the start time. RETURNING tasks.* sends the full task row back to Python in one round trip. No second query needed.
The optional task_type parameter lets you claim only a specific kind of task (e.g. only shallow scrapes or only deep scrapes).
- Save src/queue/task_queue.py.
You will not see a change yet. The next chunk adds the remaining helper functions, then you will test the full lifecycle.
Add status transitions and rate limiting
After a worker finishes processing a task, it needs to mark the job as completed or failed. You also need a rate limiter that spaces out HTTP requests to avoid overwhelming the target server.
- In src/queue/task_queue.py, add the complete_task() and fail_task() functions below claim_task():
def complete_task(task_id):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE tasks
SET status = 'completed', completed_at = CURRENT_TIMESTAMP
WHERE id = %s
""",
(task_id,),
)
conn.commit()
finally:
conn.close()
def fail_task(task_id, error_message):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE tasks
SET status = 'failed', completed_at = CURRENT_TIMESTAMP, error = %s
WHERE id = %s
""",
(error_message, task_id),
)
conn.commit()
finally:
conn.close()What do these functions do?
Both functions transition a task out of in_progress status. complete_task() marks the job as completed and timestamps when it finished.
The fail_task() function does the same but also stores an error message. This lets you inspect failures later without losing context about what went wrong.
You will not see a change yet. The next block adds the rate limiter and stats helper that complete the module.
- Still in src/queue/task_queue.py, add the wait_for_rate_limit() and get_queue_stats() functions below fail_task():
def wait_for_rate_limit():
global _last_request_time
elapsed = time.time() - _last_request_time
if elapsed < RATE_LIMIT_SECONDS:
time.sleep(RATE_LIMIT_SECONDS - elapsed)
_last_request_time = time.time()
def get_queue_stats():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT status, task_type, count(*) as count
FROM tasks
GROUP BY status, task_type
ORDER BY status, task_type
"""
)
return cur.fetchall()
finally:
conn.close()What do these helpers do?
The wait_for_rate_limit() function enforces a minimum gap of RATE_LIMIT_SECONDS (2 seconds) between HTTP requests. It checks how much time has passed since the last request and sleeps for the remainder if needed. This prevents your scraper from overwhelming the target server.
The get_queue_stats() function returns a summary of all tasks grouped by status and type. You will use this at the end of a pipeline run to see how many jobs completed, failed, or are still pending.
- Save src/queue/task_queue.py.
- Test the full task lifecycle by running:
uv run python -c "
from src.queue.task_queue import enqueue_task, claim_task, complete_task, get_queue_stats
tid = enqueue_task('shallow_scrape', {'url': 'http://test.example/page-1'}, priority=10)
print(f'Enqueued task ID: {tid}')
task = claim_task()
print(f'Claimed task {task[\"id\"]}, status: {task[\"status\"]}')
complete_task(task['id'])
print('Task completed.')
for row in get_queue_stats():
print(f' {row[\"status\"]:12} | {row[\"task_type\"]:15} | {row[\"count\"]}')
"You should see output showing the task being enqueued, claimed, completed, and a stats line showing 1 completed shallow_scrape task.
Seeing a connection refused error?
Make sure your PostgreSQL container is still running. Check with:
docker compose psIf the db service is not running, start it again with docker compose up -d. Also confirm you ran the schema creation step from Step 2. The tasks table must exist before the queue engine can insert rows.
Help me debug my PostgreSQL connection
Why use PostgreSQL as a task queue instead of Redis or RabbitMQ?
Dedicated message brokers add infrastructure complexity. For a pipeline with a single worker, PostgreSQL handles both data storage and job coordination in one service. The SKIP LOCKED pattern gives you the same atomicity guarantees that a dedicated broker would provide.
If you later need thousands of concurrent workers or sub-millisecond dispatch, you can swap in a broker. For batch ETL pipelines, PostgreSQL is more than enough.
✔️ Awesome, I've got everything!
Great. Double check you have saved src/queue/task_queue.py and move on to the next step.
ⓧ I'd like to double check the full code
import json
import time
from src.db.connection import get_connection
RATE_LIMIT_SECONDS = 2.0
_last_request_time = 0.0
def enqueue_task(task_type, payload, priority=0, domain="books.toscrape.com"):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO tasks (task_type, payload, priority, domain)
VALUES (%s, %s, %s, %s)
ON CONFLICT ((payload->>'url'))
WHERE status IN ('pending', 'in_progress')
DO NOTHING
RETURNING id
""",
(task_type, json.dumps(payload), priority, domain),
)
result = cur.fetchone()
conn.commit()
return result["id"] if result else None
finally:
conn.close()
def claim_task(task_type=None):
conn = get_connection()
try:
with conn.cursor() as cur:
if task_type:
cur.execute(
"""
WITH next_task AS (
SELECT id
FROM tasks
WHERE status = 'pending'
AND task_type = %s
ORDER BY priority DESC, created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE tasks
SET status = 'in_progress', started_at = CURRENT_TIMESTAMP
FROM next_task
WHERE tasks.id = next_task.id
RETURNING tasks.*
""",
(task_type,),
)
else:
cur.execute(
"""
WITH next_task AS (
SELECT id
FROM tasks
WHERE status = 'pending'
ORDER BY priority DESC, created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE tasks
SET status = 'in_progress', started_at = CURRENT_TIMESTAMP
FROM next_task
WHERE tasks.id = next_task.id
RETURNING tasks.*
"""
)
result = cur.fetchone()
conn.commit()
return result
finally:
conn.close()
def complete_task(task_id):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE tasks
SET status = 'completed', completed_at = CURRENT_TIMESTAMP
WHERE id = %s
""",
(task_id,),
)
conn.commit()
finally:
conn.close()
def fail_task(task_id, error_message):
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE tasks
SET status = 'failed', completed_at = CURRENT_TIMESTAMP, error = %s
WHERE id = %s
""",
(error_message, task_id),
)
conn.commit()
finally:
conn.close()
def wait_for_rate_limit():
global _last_request_time
elapsed = time.time() - _last_request_time
if elapsed < RATE_LIMIT_SECONDS:
time.sleep(RATE_LIMIT_SECONDS - elapsed)
_last_request_time = time.time()
def get_queue_stats():
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT status, task_type, count(*) as count
FROM tasks
GROUP BY status, task_type
ORDER BY status, task_type
"""
)
return cur.fetchall()
finally:
conn.close()Your task queue engine is ready to coordinate work. Next up, you will build the Playwright scrapers that consume these tasks and fill the books table with real data.
Build the Playwright Scrapers
Your task queue knows how to manage jobs, but nothing is actually fetching web pages yet. The scrapers are the workers that turn queued URLs into data stored in PostgreSQL.
In this step, you will build two Playwright scrapers that work in two passes. A shallow scraper scans listing pages to discover book URLs, and a deep scraper visits each individual book page to extract full metadata and preserve the raw HTML. You will also create a browser context layer that blocks unnecessary resources like images and fonts to keep requests fast.
In this step, get ready to:
- Build a browser context with resource-blocking middleware.
- Implement a shallow scraper that navigates listing pages and enqueues deep-scrape jobs.
- Implement a deep scraper that extracts full book metadata and preserves raw HTML.
Set up the browser context with resource blocking
Every page a browser loads includes images, stylesheets, fonts, and tracking scripts. For scraping, you only need the HTML structure. Blocking those extra resources reduces bandwidth and speeds up each page load significantly.
- Create the scraper package directory by running these commands:
mkdir src/scraper
touch src/scraper/__init__.py- Confirm the files exist by running:
ls src/scraper/You should see __init__.py in the output.
- Create a new file src/scraper/browser.py and add the following code:
from playwright.sync_api import sync_playwright, Page
# Resource types that add no value to HTML scraping
BLOCKED_RESOURCE_TYPES = {"image", "stylesheet", "font", "media"}
# Common tracking/ad domains to reject
BLOCKED_URL_PATTERNS = ["google-analytics", "doubleclick", "facebook"]
def should_block(route):
request = route.request
# Block by resource type (images, CSS, fonts, media)
if request.resource_type in BLOCKED_RESOURCE_TYPES:
route.abort()
return
# Block by URL pattern (analytics, ads)
for pattern in BLOCKED_URL_PATTERNS:
if pattern in request.url:
route.abort()
return
# Allow everything else through
route.continue_()
def create_browser_context(playwright):
browser = playwright.chromium.launch(headless=True)
context = browser.new_context()
# Intercept every network request through our filter
context.route("**/*", should_block)
return browser, contextWhat does this code do?
This module sets up a lean browser environment for scraping.
- BLOCKED_RESOURCE_TYPES lists resource categories (images, CSS, fonts, media) that Playwright will abort on sight.
- BLOCKED_URL_PATTERNS catches tracking scripts by matching URL substrings.
- should_block(route) is the request interceptor. It checks each request against both filters, aborting matches and letting everything else through via route.continue_().
- create_browser_context(playwright) launches headless Chromium and routes all requests through the should_block handler.
- Save src/scraper/browser.py.
- Verify the module loads correctly by running:
uv run python -c "from src.scraper.browser import create_browser_context; print('Browser module OK')"You should see Browser module OK printed with no errors.
Seeing a ModuleNotFoundError?
- Make sure you are running the command from the project root (the folder containing pyproject.toml).
- Check that src/scraper/__init__.py exists. Without it, Python cannot find the src.scraper package.
Help me debug my Playwright browser module import error.
Why block resources instead of using raw HTTP?
A raw HTTP library like requests would be faster, but it cannot execute JavaScript. Some sites render content dynamically. Playwright gives you a real browser that handles JavaScript-rendered pages while the resource filter keeps things lean.
✔️ Awesome, I've got everything!
Great. Make sure you have saved your file before moving on.
ⓧ I'd like to double check the full code
from playwright.sync_api import sync_playwright, Page
BLOCKED_RESOURCE_TYPES = {"image", "stylesheet", "font", "media"}
BLOCKED_URL_PATTERNS = ["google-analytics", "doubleclick", "facebook"]
def should_block(route):
request = route.request
if request.resource_type in BLOCKED_RESOURCE_TYPES:
route.abort()
return
for pattern in BLOCKED_URL_PATTERNS:
if pattern in request.url:
route.abort()
return
route.continue_()
def create_browser_context(playwright):
browser = playwright.chromium.launch(headless=True)
context = browser.new_context()
context.route("**/*", should_block)
return browser, contextImplement the shallow scraper for listing pages
The shallow scraper handles the first pass of the two-pass strategy. It visits listing pages on books.toscrape.com, extracts basic metadata from each book card, and enqueues a deep-scrape job for every book it discovers.
- Create a new file src/scraper/shallow.py and add the following code:
from src.queue.task_queue import enqueue_task
# Base URL for all book pages
BASE_URL = "https://books.toscrape.com"
# Map star-rating class names to numeric strings
RATING_MAP = {
"One": "1",
"Two": "2",
"Three": "3",
"Four": "4",
"Five": "5",
}
def scrape_listing_page(page, url):
page.goto(url, wait_until="domcontentloaded")
books = []
# Each book card lives inside an article.product_pod element
articles = page.locator("article.product_pod").all()
for article in articles:
# Extract title and relative link from the heading anchor
link_el = article.locator("h3 a")
title = link_el.get_attribute("title")
relative_url = link_el.get_attribute("href")
# Normalize relative URLs into absolute book URLs
if relative_url.startswith("catalogue/"):
book_url = f"{BASE_URL}/{relative_url}"
elif relative_url.startswith("../"):
book_url = f"{BASE_URL}/catalogue/{relative_url.lstrip('../')}"
else:
book_url = f"{BASE_URL}/catalogue/{relative_url}"
# Extract price from the price paragraph
price_el = article.locator("p.price_color")
price = price_el.text_content().strip()
# Extract star rating from the class name
rating_el = article.locator("p.star-rating")
rating_class = rating_el.get_attribute("class")
rating_word = rating_class.replace("star-rating ", "")
rating = RATING_MAP.get(rating_word, "0")
books.append({
"url": book_url,
"title": title,
"price": price,
"rating": rating,
})
# Enqueue a deep-scrape job for this book
enqueue_task(
task_type="deep_scrape",
payload={"url": book_url, "title": title, "price": price, "rating": rating},
priority=5,
)
return booksWhat does this code do?
- scrape_listing_page(page, url) navigates to a listing page and finds every article.product_pod element on the page.
- For each book card, it extracts the title, price, star rating, and builds an absolute URL from the relative link.
- The URL normalization handles two formats that books.toscrape.com uses: paths starting with catalogue/ and paths starting with ../.
- Each discovered book gets enqueued as a deep_scrape task at priority 5. This is lower than the listing pages (priority 10), so the pipeline finishes discovering books before deep-scraping them.
- Save src/scraper/shallow.py.
- Verify the module imports correctly by running:
uv run python -c "from src.scraper.shallow import scrape_listing_page; print('Shallow scraper OK')"You should see Shallow scraper OK printed with no errors.
Seeing an import error for task_queue?
- Confirm that src/queue/__init__.py exists. Python needs this file to recognize src.queue as a package.
- Make sure your PostgreSQL container is still running. The task_queue module imports get_connection which attempts to connect at import time if called.
Help me debug my shallow scraper import.
✔️ Awesome, I've got everything!
Great. Make sure you have saved your file before moving on.
ⓧ I'd like to double check the full code
from src.queue.task_queue import enqueue_task
BASE_URL = "https://books.toscrape.com"
RATING_MAP = {
"One": "1",
"Two": "2",
"Three": "3",
"Four": "4",
"Five": "5",
}
def scrape_listing_page(page, url):
page.goto(url, wait_until="domcontentloaded")
books = []
articles = page.locator("article.product_pod").all()
for article in articles:
link_el = article.locator("h3 a")
title = link_el.get_attribute("title")
relative_url = link_el.get_attribute("href")
if relative_url.startswith("catalogue/"):
book_url = f"{BASE_URL}/{relative_url}"
elif relative_url.startswith("../"):
book_url = f"{BASE_URL}/catalogue/{relative_url.lstrip('../')}"
else:
book_url = f"{BASE_URL}/catalogue/{relative_url}"
price_el = article.locator("p.price_color")
price = price_el.text_content().strip()
rating_el = article.locator("p.star-rating")
rating_class = rating_el.get_attribute("class")
rating_word = rating_class.replace("star-rating ", "")
rating = RATING_MAP.get(rating_word, "0")
books.append({
"url": book_url,
"title": title,
"price": price,
"rating": rating,
})
enqueue_task(
task_type="deep_scrape",
payload={"url": book_url, "title": title, "price": price, "rating": rating},
priority=5,
)
return booksImplement the deep scraper for detail pages
The deep scraper handles the second pass. It visits individual book pages to extract the full metadata that listing pages do not include: the UPC, description, category, and availability. It also captures the entire raw HTML so the pipeline preserves the original source for future enrichment stages.
- Create a new file src/scraper/deep.py and add the following code:
import psycopg2
from src.db.connection import get_connection
def scrape_book_detail(page, url, metadata):
page.goto(url, wait_until="domcontentloaded")
# Capture the full page HTML for raw preservation
raw_html = page.content()
# Extract category from the breadcrumb navigation
breadcrumb_links = page.locator("ul.breadcrumb li a").all()
category = ""
if len(breadcrumb_links) >= 3:
category = breadcrumb_links[2].text_content().strip()
# Extract description from the paragraph after #product_description
description_el = page.locator("#product_description ~ p")
description = ""
if description_el.count() > 0:
description = description_el.first.text_content().strip()
# Extract UPC and availability from the product info table
table_rows = page.locator("table.table-striped tr").all()
upc = ""
availability = ""
for row in table_rows:
header = row.locator("th").text_content().strip()
value = row.locator("td").text_content().strip()
if header == "UPC":
upc = value
elif header == "Availability":
availability = value
# Upsert into the books table with all extracted fields
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO books (url, title, price, rating, category, availability, upc, description, raw_html)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (url) DO UPDATE SET
title = EXCLUDED.title,
price = EXCLUDED.price,
rating = EXCLUDED.rating,
category = EXCLUDED.category,
availability = EXCLUDED.availability,
upc = EXCLUDED.upc,
description = EXCLUDED.description,
raw_html = EXCLUDED.raw_html,
scraped_at = CURRENT_TIMESTAMP
""",
(
url,
metadata.get("title", ""),
metadata.get("price", ""),
metadata.get("rating", ""),
category,
availability,
upc,
description,
raw_html,
),
)
conn.commit()
finally:
conn.close()What does this code do?
- page.content() captures the full page HTML as a string. This raw HTML is stored alongside structured fields so no data is lost if you later decide to extract additional fields.
- The category comes from the third breadcrumb link (Home > Books > Category > Title).
- The #product_description ~ p CSS selector targets the paragraph immediately after the description heading.
- The ON CONFLICT (url) DO UPDATE clause performs an upsert. If a book URL already exists, it updates the row instead of failing. This makes re-runs safe.
- Save src/scraper/deep.py.
- Verify that all three scraper modules import successfully by running:
uv run python -c "from src.scraper.browser import create_browser_context; from src.scraper.shallow import scrape_listing_page; from src.scraper.deep import scrape_book_detail; print('All scrapers imported successfully')"You should see All scrapers imported successfully with no errors.
Seeing a connection error from psycopg2?
- The deep scraper imports get_connection at module level, but it only connects when scrape_book_detail() is called. If you see a connection error during import, make sure your Docker Compose stack is running.
- Run docker compose ps and confirm the db service shows as healthy.
Help me fix my deep scraper import issue.
✔️ Awesome, I've got everything!
Great. Make sure you have saved your file before moving on.
ⓧ I'd like to double check the full code
import psycopg2
from src.db.connection import get_connection
def scrape_book_detail(page, url, metadata):
page.goto(url, wait_until="domcontentloaded")
raw_html = page.content()
breadcrumb_links = page.locator("ul.breadcrumb li a").all()
category = ""
if len(breadcrumb_links) >= 3:
category = breadcrumb_links[2].text_content().strip()
description_el = page.locator("#product_description ~ p")
description = ""
if description_el.count() > 0:
description = description_el.first.text_content().strip()
table_rows = page.locator("table.table-striped tr").all()
upc = ""
availability = ""
for row in table_rows:
header = row.locator("th").text_content().strip()
value = row.locator("td").text_content().strip()
if header == "UPC":
upc = value
elif header == "Availability":
availability = value
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO books (url, title, price, rating, category, availability, upc, description, raw_html)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (url) DO UPDATE SET
title = EXCLUDED.title,
price = EXCLUDED.price,
rating = EXCLUDED.rating,
category = EXCLUDED.category,
availability = EXCLUDED.availability,
upc = EXCLUDED.upc,
description = EXCLUDED.description,
raw_html = EXCLUDED.raw_html,
scraped_at = CURRENT_TIMESTAMP
""",
(
url,
metadata.get("title", ""),
metadata.get("price", ""),
metadata.get("rating", ""),
category,
availability,
upc,
description,
raw_html,
),
)
conn.commit()
finally:
conn.close()Your scrapers are ready to go. In the next step, you will build the orchestrator that seeds listing-page jobs into the queue, launches a worker loop, and runs the full pipeline end-to-end against books.toscrape.com.
Run the Pipeline End-to-End
You have a task queue, a shallow scraper, and a deep scraper. All the individual pieces are built and tested.
Now you need an orchestrator to tie them together. The orchestrator seeds the queue with listing-page URLs, then runs a worker loop that claims tasks and dispatches them to the right scraper based on task_type.
In this step, get ready to:
- Build the pipeline orchestrator that seeds jobs and runs a dispatch loop.
- Run the full pipeline against books.toscrape.com (3 pages, 60 books).
- Verify that all 60 books landed in PostgreSQL with raw HTML preserved.
Build the orchestrator
The orchestrator is the entry point for the entire pipeline. It creates the schema, seeds initial jobs, launches the browser, and loops until the queue is empty.
- Create a new file called src/main.py inside your src folder.
- Add the imports and configuration by pasting this code:
import time
from playwright.sync_api import sync_playwright
from src.db.schema import create_schema
from src.queue.task_queue import (
enqueue_task,
claim_task,
complete_task,
fail_task,
wait_for_rate_limit,
get_queue_stats,
)
from src.scraper.browser import create_browser_context
from src.scraper.shallow import scrape_listing_page
from src.scraper.deep import scrape_book_detail
BASE_URL = "https://books.toscrape.com"
PAGES_TO_SCRAPE = 3What does this code do?
The imports pull in every module you built across the previous steps: the schema creator, the task queue functions, the browser context factory, and both scrapers.
The two constants configure the target site and how many listing pages to scrape. Three pages at 20 books each gives you 60 books total.
- Below the constants, add the seed_listing_jobs() function:
def seed_listing_jobs():
print(f"Seeding {PAGES_TO_SCRAPE} listing page jobs...")
for i in range(1, PAGES_TO_SCRAPE + 1):
url = f"{BASE_URL}/catalogue/page-{i}.html"
enqueue_task(
task_type="shallow_scrape",
payload={"url": url, "page_number": i},
priority=10,
)
print("Listing jobs seeded.")What does this code do?
This function creates the initial work for the pipeline. It enqueues one shallow_scrape task per listing page, using priority 10 so listing pages get processed before deep-scrape jobs (which have priority 5).
The higher priority ensures the pipeline discovers all books first, then scrapes their details. This avoids running out of deep-scrape jobs while listing pages remain unprocessed.
- Below seed_listing_jobs(), add the run_worker() function:
def run_worker():
with sync_playwright() as pw:
browser, context = create_browser_context(pw)
page = context.new_page()
try:
while True:
task = claim_task()
if task is None:
print("No more tasks. Pipeline complete.")
break
task_id = task["id"]
task_type = task["task_type"]
payload = task["payload"]
url = payload["url"]
print(f"Processing [{task_type}]: {url}")
wait_for_rate_limit()
try:
if task_type == "shallow_scrape":
books = scrape_listing_page(page, url)
print(f" Found {len(books)} books on listing page.")
elif task_type == "deep_scrape":
scrape_book_detail(page, url, payload)
print(f" Stored book detail.")
else:
print(f" Unknown task type: {task_type}")
complete_task(task_id)
except Exception as e:
print(f" Error: {e}")
fail_task(task_id, str(e))
finally:
context.close()
browser.close()What does this code do?
The worker loop is the heart of the pipeline. It opens one Playwright browser session and reuses it across all tasks.
- claim_task() atomically grabs the next pending job using SKIP LOCKED. When it returns None, the queue is empty and the loop exits.
- The dispatcher checks task_type and routes to either scrape_listing_page() or scrape_book_detail().
- If a task fails, fail_task() records the error message in the database so you can debug later.
The finally block guarantees the browser closes cleanly even if an unhandled exception occurs.
- Below run_worker(), add the main() function and entry point:
def main():
create_schema()
seed_listing_jobs()
print("\nStarting worker loop...\n")
run_worker()
print("\n--- Pipeline Stats ---")
stats = get_queue_stats()
for row in stats:
print(f" {row['status']:12} | {row['task_type']:15} | {row['count']}")
from src.db.connection import get_connection
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT count(*) as total, count(DISTINCT url) as unique_urls, count(raw_html) as with_html FROM books")
result = cur.fetchone()
print(f"\n--- Books Table ---")
print(f" Total: {result['total']}")
print(f" Unique URLs: {result['unique_urls']}")
print(f" With raw HTML: {result['with_html']}")
finally:
conn.close()
if __name__ == "__main__":
main()What does this code do?
The main() function orchestrates the entire pipeline in sequence: ensure the schema exists, seed the initial jobs, run the worker until the queue is empty, then print a summary.
The verification query at the end counts total books, unique URLs, and non-null raw_html entries. This gives you a quick sanity check that the pipeline completed without duplicates or missing data.
- Save src/main.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved the file before running the pipeline.
ⓧ I'd like to double check the full code
import time
from playwright.sync_api import sync_playwright
from src.db.schema import create_schema
from src.queue.task_queue import (
enqueue_task,
claim_task,
complete_task,
fail_task,
wait_for_rate_limit,
get_queue_stats,
)
from src.scraper.browser import create_browser_context
from src.scraper.shallow import scrape_listing_page
from src.scraper.deep import scrape_book_detail
BASE_URL = "https://books.toscrape.com"
PAGES_TO_SCRAPE = 3
def seed_listing_jobs():
print(f"Seeding {PAGES_TO_SCRAPE} listing page jobs...")
for i in range(1, PAGES_TO_SCRAPE + 1):
url = f"{BASE_URL}/catalogue/page-{i}.html"
enqueue_task(
task_type="shallow_scrape",
payload={"url": url, "page_number": i},
priority=10,
)
print("Listing jobs seeded.")
def run_worker():
with sync_playwright() as pw:
browser, context = create_browser_context(pw)
page = context.new_page()
try:
while True:
task = claim_task()
if task is None:
print("No more tasks. Pipeline complete.")
break
task_id = task["id"]
task_type = task["task_type"]
payload = task["payload"]
url = payload["url"]
print(f"Processing [{task_type}]: {url}")
wait_for_rate_limit()
try:
if task_type == "shallow_scrape":
books = scrape_listing_page(page, url)
print(f" Found {len(books)} books on listing page.")
elif task_type == "deep_scrape":
scrape_book_detail(page, url, payload)
print(f" Stored book detail.")
else:
print(f" Unknown task type: {task_type}")
complete_task(task_id)
except Exception as e:
print(f" Error: {e}")
fail_task(task_id, str(e))
finally:
context.close()
browser.close()
def main():
create_schema()
seed_listing_jobs()
print("\nStarting worker loop...\n")
run_worker()
print("\n--- Pipeline Stats ---")
stats = get_queue_stats()
for row in stats:
print(f" {row['status']:12} | {row['task_type']:15} | {row['count']}")
from src.db.connection import get_connection
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT count(*) as total, count(DISTINCT url) as unique_urls, count(raw_html) as with_html FROM books")
result = cur.fetchone()
print(f"\n--- Books Table ---")
print(f" Total: {result['total']}")
print(f" Unique URLs: {result['unique_urls']}")
print(f" With raw HTML: {result['with_html']}")
finally:
conn.close()
if __name__ == "__main__":
main()Run the pipeline against books.toscrape.com
With the orchestrator built, it is time to let the pipeline loose on the live site. The worker will scrape 3 listing pages, discover 60 books, then scrape each book's detail page.
How long will this take?
The rate limiter enforces a 2-second delay between requests. With 3 listing pages plus 60 detail pages, expect the pipeline to run for approximately 2 minutes.
- Make sure your PostgreSQL container is still running. Confirm by running:
docker compose ps- You should see the db service with status healthy. If it is not running, start it by running:
docker compose up -d- Launch the pipeline by running:
uv run python -m src.mainYou should see output scrolling through the terminal as the pipeline processes each task.
What should I see?
The pipeline prints progress for each task it claims. First you will see the three listing pages being processed, each reporting 20 books found. Then the 60 deep-scrape jobs process one by one, each printing "Stored book detail." as it completes.
At the very end, the summary tables print. The Pipeline Stats show all tasks as completed. The Books Table shows 60 total, 60 unique URLs, and 60 with raw HTML.
Seeing connection refused errors?
- Verify your PostgreSQL container is healthy with docker compose ps. If it shows "starting", wait a few seconds and try again.
- If you see a Playwright browser launch error, make sure you ran playwright install in your project's virtual environment during Step 1 setup.
- If a few tasks fail but most succeed, that is normal for network flakiness. The failed tasks are recorded in the database with their error messages.
Ask for help debugging your pipeline run.
Verify the results in PostgreSQL
The pipeline already printed a summary, but you can also query PostgreSQL directly to confirm the data landed correctly.
- Connect to the database by running:
docker compose exec db psql -U scraper -d bookpipeline- Run this verification query:
SELECT count(*) AS total, count(DISTINCT url) AS unique_urls, count(raw_html) AS with_html FROM books;- You should see 60 for all three columns: 60 total books, 60 unique URLs, and 60 non-null raw_html entries.
- Exit psql by typing:
\qWhy 60/60/60 matters
Three matching numbers prove three things at once. 60 total means the pipeline processed every book it discovered. 60 unique URLs means the ON CONFLICT deduplication worked correctly. 60 non-null raw_html entries mean the deep scraper preserved the full page content for every book.
This raw HTML is what makes the pipeline production-grade. Future enrichment stages can re-parse the HTML without re-scraping the site.
Seeing fewer than 60 books?
- Check the tasks table for failed jobs: SELECT count(*) FROM tasks WHERE status = 'failed';. If some failed, inspect their error column for details.
- If you previously ran the pipeline and want a fresh start, truncate both tables: TRUNCATE tasks, books RESTART IDENTITY; then run the pipeline again.
- Network timeouts can cause individual tasks to fail. Re-running the pipeline is safe because ON CONFLICT DO NOTHING prevents duplicate task creation.
Help me figure out why my pipeline scraped fewer than 60 books.
Secret mission
Weighted Round-Robin Category Scheduling
Your pipeline processes books in whatever order the queue serves them. But what if some categories matter more than others? In this mission, you'll build a weighted round-robin scheduler that balances scrape effort across categories while preventing deep-scrape starvation.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- PostgreSQL container with persistent volume (pgdata).
- Playwright browser binaries (Chromium).
- book-scraper project folder on your Desktop.
Keep everything running
No action needed. Choose this if you're still actively building or plan to extend this pipeline with additional stages.
- Your PostgreSQL container keeps running in the background. Run docker compose up -d from the book-scraper folder if you previously stopped it.
- Your scraped book data remains in the database, ready for a transformation stage.
- Playwright browser binaries remain cached on your system for future scraping runs.
Pause - I'll come back to this later
Shut down running processes to free up system resources while keeping your data intact.
- Stop the PostgreSQL container by running this from your book-scraper folder:
docker compose stop- This preserves all data in the pgdata volume. When you're ready to resume, run docker compose up -d to start the container again.
- Your project files and Playwright binaries stay on disk unchanged.
Delete - I don't want to use this again
Remove all project resources and start fresh. This is irreversible.
- Remove the PostgreSQL container and its persistent volume by running this from your book-scraper folder:
docker compose down -v- This removes the container, the pgdata volume, and all scraped book data stored in PostgreSQL.
- Remove the Playwright browser binaries by running:
playwright uninstall- Delete the project folder by running:
rm -rf ~/Desktop/book-scraper- This removes all Python source files, the docker-compose.yml, and the pyproject.toml.
Nice Work!
Nice Work!
You just built a production-shaped scraping pipeline backed by a transactional task queue. Your system scrapes books.toscrape.com across two passes, preserves every byte of raw HTML, and guarantees no duplicate work through atomic job claiming.
You've learned how to:
- Build a PostgreSQL-backed task queue using FOR UPDATE SKIP LOCKED that atomically claims jobs without duplicate processing.
- Create a two-pass Playwright scraper (shallow listings + deep details) that preserves raw HTML for future enrichment.
- Containerize the database layer with Docker Compose, ready for additional pipeline stages.
- Secret Mission: Implement weighted round-robin scheduling that balances scrape effort across categories while preventing deep-scrape starvation.
Ready to quiz yourself?