Playwright scrapers handle JavaScript-rendered pages, SPAs, and complex interactions that static HTTP clients can’t touch. But they’re equally susceptible to silent failures: selector drift, DOM restructuring, and data quality degradation.

This tutorial shows two ways to integrate ScraperGuard with Playwright scripts: the observe() context manager for multi-page workflows, and the capture_page() one-shot function for simple captures.

Install

pip install -e ".[playwright]"
playwright install chromium

Method 1: The `observe()` Context Manager

For scrapers that navigate pages, extract data, and then want ScraperGuard to analyze everything:

import asyncio
from playwright.async_api import async_playwright
from scraperguard.integrations.playwright import observe
from scraperguard.storage.sqlite import SQLiteBackend
from scraperguard.schema import BaseSchema, validators

# Define your schema
class ProductSchema(BaseSchema):
    title: str
    price: float = validators.range(min=0.01)
    in_stock: bool

async def scrape_with_monitoring():
    storage = SQLiteBackend("scraperguard.db")

    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        # Wrap your scraping logic with observe()
        async with observe(
            page,
            storage=storage,
            run_id="run-001",
            schema=ProductSchema,
            selectors=[".price-tag", ".product-title", ".stock-badge"]
        ) as observer:
            # Your normal Playwright scraping code — unchanged
            await page.goto("https://shop.example.com/products")
            await page.wait_for_selector(".product-grid")

            # Extract data however you normally would
            products = await page.evaluate("""
                () => Array.from(document.querySelectorAll('.product-card')).map(el => ({
                    title: el.querySelector('.product-title')?.textContent?.trim(),
                    price: parseFloat(el.querySelector('.price-tag')?.textContent?.replace('$', '')),
                    in_stock: el.querySelector('.stock-badge')?.textContent?.includes('In Stock')
                }))
            """)

            # Tell ScraperGuard what you extracted
            observer.set_items(products)

        # When the context manager exits, ScraperGuard automatically:
        # 1. Captures the current page HTML via page.content()
        # 2. Normalizes and fingerprints the DOM
        # 3. Validates items against ProductSchema
        # 4. Compares against previous snapshot (if exists)
        # 5. Detects null drift against baseline
        # 6. Classifies any failures
        # 7. Computes health score
        # 8. Fires alerts if thresholds breached
        # 9. Stores everything in SQLite

        await browser.close()

asyncio.run(scrape_with_monitoring())

What Happens Inside `observe()`

The context manager follows the observer pattern — it wraps your code without modifying it:

On entry (__aenter__): Returns an observer object. Your scraping code runs normally inside the async with block.

During execution: You call observer.set_items(data) to tell ScraperGuard what you extracted.

On exit (__aexit__): ScraperGuard captures the page state via page.content() and runs the full analysis pipeline. If your code raises an exception, ScraperGuard still captures what it can and re-raises the exception — it never swallows errors.

The key property: your scraping logic is unchanged. The observe() wrapper is additive. Remove it, and your scraper works exactly as before.

Inside the PageObserver Class

The PageObserver class (integrations/playwright/observer.py) manages the full lifecycle:

class PageObserver:
    """Observes a Playwright page and runs analysis on context exit."""

    def __init__(
        self,
        page: object,
        storage: StorageBackend,
        run_id: str,
        config: ScraperGuardConfig | None = None,
        selectors: list[str] | None = None,
        schema: type[BaseSchema] | None = None,
    ) -> None:
        self.page = page
        self.storage = storage
        self.run_id = run_id
        self.config = config or load_config()
        self.selectors = selectors or []
        self.schema = schema
        self._items: list[dict] = []
        self._url: str | None = None
        self._health_report: HealthReport | None = None

    def set_items(self, items: list[dict]) -> None:
        """Store extracted items for validation."""
        self._items = items

    def set_url(self, url: str) -> None:
        """Override URL (useful for SPAs where page.url might not reflect content)."""
        self._url = url

    def get_health_report(self) -> HealthReport | None:
        """Retrieve the health report after context exit."""
        return self._health_report

    async def __aenter__(self) -> "PageObserver":
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
        """Run full analysis pipeline. Never suppresses exceptions."""
        try:
            html = await self.page.content()
            url = self._url or self.page.url

            # Run the exact same pipeline as the Scrapy integration:
            # snapshot → validate → drift → diff → classify → score → alert
            self._health_report = await self._analyze(url, html)
        except Exception as e:
            logger.warning(f"ScraperGuard analysis failed: {e}")
            # Never crash the user's scraper

        return False  # Never suppress user exceptions

The key guarantee: __aexit__ returns False, meaning if your scraping code raises an exception, it propagates normally. ScraperGuard’s own errors are caught internally and logged as warnings.

Method 2: The `capture_page()` One-Shot Function

For simple scenarios where you’ve already navigated and extracted data, and just want a health check:

from scraperguard.integrations.playwright import capture_page

async def quick_check():
    storage = SQLiteBackend("scraperguard.db")

    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        await page.goto("https://shop.example.com/products")

        # Extract your data
        items = await extract_products(page)

        # One-shot health check
        report = await capture_page(
            page,
            storage=storage,
            run_id="run-002",
            items=items,
            schema=ProductSchema,
            selectors=[".price-tag", ".product-title"]
        )

        if report:
            print(f"Health: {report.overall_score}/100 ({report.status})")
            if report.overall_score < 50:
                print("CRITICAL: Scraper health degraded!")
                for failure in report.classifications:
                    print(f"  [{failure.severity}] {failure.failure_type}: {failure.recommendation}")

        await browser.close()

capture_page() returns a HealthReport object (or None if analysis couldn’t complete). It runs the same pipeline as observe() but in a single function call.

Using Health Reports for Conditional Retries

The HealthReport from observer.get_health_report() is useful for Playwright-specific retry logic — if the page looks wrong, you can re-navigate or switch to a different rendering strategy:

async def scrape_with_retry(page, url, storage, schema):
    """Scrape a page and retry with full JS rendering if health is poor."""
    for attempt in range(3):
        async with observe(page, storage=storage,
                           run_id=f"run-{attempt}",
                           schema=schema,
                           selectors=[".price", ".title"]) as observer:
            await page.goto(url, wait_until="domcontentloaded")
            items = await extract_products(page)
            observer.set_items(items)

        report = observer.get_health_report()
        if report is None or report.overall_score >= 70:
            return items  # Good enough

        # Check what went wrong
        for c in (report.classifications or []):
            if c.failure_type.value == "js_challenge":
                # Page needs full JS rendering — wait longer
                await page.goto(url, wait_until="networkidle")
                continue
            if c.failure_type.value == "captcha_injection":
                # CAPTCHA detected — no point retrying
                raise RuntimeError(f"CAPTCHA on {url}: {c.evidence}")

        # Generic degraded health — wait and retry
        await page.wait_for_timeout(2000 * (attempt + 1))

    return items  # Return best effort after retries

For the full HealthReport structure (components, classifications, drift events, export format), see the production architecture guide.

Monitoring Multiple Pages in One Session

For scrapers that visit multiple pages, use observe() per page or accumulate items and capture once:

async def scrape_catalog():
    storage = SQLiteBackend("scraperguard.db")
    all_items = []

    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        # Scrape multiple pages
        for page_num in range(1, 11):
            url = f"https://shop.example.com/products?page={page_num}"
            await page.goto(url)
            items = await extract_products(page)
            all_items.extend(items)

        # Analyze the complete batch on the last page
        report = await capture_page(
            page,
            storage=storage,
            run_id=f"catalog-{datetime.now().isoformat()}",
            items=all_items,
            schema=ProductSchema,
            selectors=[".price-tag", ".product-title"]
        )

        print(f"Scraped {len(all_items)} products. Health: {report.overall_score}/100")

        await browser.close()

Per-Page Health Monitoring

For scrapers where each page has independent content, you can monitor health per page:

async def scrape_with_per_page_monitoring():
    storage = SQLiteBackend("scraperguard.db")

    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        urls = [
            "https://shop.example.com/electronics",
            "https://shop.example.com/clothing",
            "https://shop.example.com/books",
        ]

        results = []
        for url in urls:
            async with observe(
                page, storage=storage,
                run_id=f"monitor-{url.split('/')[-1]}",
                schema=ProductSchema,
                selectors=[".price", ".title", ".stock"]
            ) as observer:
                await page.goto(url)
                items = await extract_products(page)
                observer.set_items(items)

            report = observer.get_health_report()
            results.append({
                "url": url,
                "score": report.overall_score if report else None,
                "status": report.status if report else "unknown",
            })

        # Summary across all pages
        print("\n--- Health Summary ---")
        for r in results:
            emoji = "✓" if r["status"] == "healthy" else "✗"
            print(f"  {emoji} {r['url']}: {r['score']}/100 ({r['status']})")

        critical = [r for r in results if r["status"] == "critical"]
        if critical:
            print(f"\n⚠ {len(critical)} pages in critical state!")

        await browser.close()

Combining with Alerting

Wire up Slack notifications for critical failures:

alerts:
  slack:
    enabled: true
    webhook: ${SLACK_WEBHOOK_URL}
  thresholds:
    health_score: 60
    selector_break: warning

When a Playwright scraper’s health drops below 60 or a selector breaks, you’ll get a Slack message with the diagnosis.

Using the API to Monitor Playwright Runs

After your Playwright scraper has run several times with ScraperGuard:

# See all runs and their health scores
scraperguard serve --port 8000
# Or use the CLI
scraperguard report --url https://shop.example.com/products --format json

The REST API exposes all historical data: runs, snapshots, validation results, drift events, and selector tracking. Build a monitoring dashboard or pipe the data into your existing observability stack.

Error Handling

ScraperGuard is designed to never crash your scraper. All analysis is wrapped in try/except internally:

# If ScraperGuard encounters an error during analysis,
# it logs the error and continues. Your scraper is unaffected.

async with observe(page, storage=storage, run_id="run-003", schema=ProductSchema) as observer:
    await page.goto(url)
    items = await extract(page)
    observer.set_items(items)

    # Even if the page HTML is malformed, or storage is full,
    # or the schema has a bug — the context manager exits cleanly
    # and your items are still valid.

If the Playwright library isn’t installed, importing the integration raises a helpful error:

# Without playwright installed:
from scraperguard.integrations.playwright import observe
# → ImportError: Playwright integration requires playwright.
#   Install it with: pip install scraperguard[playwright]

Using Both Scrapy and Playwright

For hybrid scrapers that use Scrapy for most pages but Playwright for JavaScript-heavy pages:

import scrapy
from scrapy_playwright.page import PageMethod
from scraperguard.integrations.playwright import capture_page

class HybridSpider(scrapy.Spider):
    name = "hybrid"

    def start_requests(self):
        # Regular pages — Scrapy middleware handles monitoring
        yield scrapy.Request("https://example.com/simple-page")

        # JS-rendered pages — use Playwright with explicit monitoring
        yield scrapy.Request(
            "https://example.com/spa-page",
            meta={"playwright": True,
                  "playwright_page_methods": [
                      PageMethod("wait_for_selector", ".product-grid"),
                  ]},
        )

    async def parse(self, response):
        if response.meta.get("playwright"):
            # Playwright page available — use ScraperGuard's Playwright observer
            page = response.meta["playwright_page"]
            items = await self._extract_from_page(page)

            report = await capture_page(
                page, storage=self.storage,
                run_id=self.crawler.stats.get_value("scraperguard_run_id"),
                items=items, schema=ProductSchema,
            )
            if report and report.overall_score < 50:
                self.logger.critical(f"SPA page health critical: {report.overall_score}")

            for item in items:
                yield item
        else:
            # Regular response — ScraperGuard middleware handles this automatically
            yield {
                "title": response.css(".title::text").get(),
                "price": float(response.css(".price::text").get("0")),
            }

This completes the Scraper Reliability & Observability series. For the complete architecture and design philosophy, see From Scripts to Systems: Production Scraping Architecture. For the case study of how ScraperGuard was built, see Building ScraperGuard: Architecture Decisions and Tradeoffs.

On This Page

Monitoring Playwright Scrapers with ScraperGuard

Install

Method 1: The `observe()` Context Manager

What Happens Inside `observe()`

Inside the PageObserver Class

Method 2: The `capture_page()` One-Shot Function

Using Health Reports for Conditional Retries

Monitoring Multiple Pages in One Session

Per-Page Health Monitoring

Combining with Alerting

Using the API to Monitor Playwright Runs

Error Handling

Using Both Scrapy and Playwright

Comments

Install

Method 1: The observe() Context Manager

What Happens Inside observe()

Inside the PageObserver Class

Method 2: The capture_page() One-Shot Function

Using Health Reports for Conditional Retries

Monitoring Multiple Pages in One Session

Per-Page Health Monitoring

Combining with Alerting

Using the API to Monitor Playwright Runs

Error Handling

Using Both Scrapy and Playwright

Comments

Method 1: The `observe()` Context Manager

What Happens Inside `observe()`

Method 2: The `capture_page()` One-Shot Function