Every web scraping tutorial teaches you how to extract data. Nobody teaches you how to verify it.

This tutorial walks you through building a data validation layer for scraping pipelines — starting with basic type checking, progressing through batch statistics, and ending with statistical drift detection that catches subtle degradation over time. Every concept is illustrated with working code and connected to how ScraperGuard implements it internally.

Why Scrapers Need Validation

Consider this item extracted from a product page:

item = {
    "title": "Wireless Bluetooth Headphones",
    "price": 79.99,
    "availability": True,
    "rating": 4.3
}

This looks fine. Now consider these items the scraper might produce after a site change:

# Selector broke — price returns None
{"title": "Headphones", "price": None, "availability": True, "rating": 4.3}

# Wrong type — price extracted as string including currency symbol
{"title": "Headphones", "price": "$79.99", "availability": True, "rating": 4.3}

# CAPTCHA page — extracted title is the challenge text
{"title": "Please complete the CAPTCHA to continue", "price": None, "availability": None, "rating": None}

All three flow through a typical pipeline into your database without raising any exception. The spider ran, it returned items, it wrote to storage. No errors in the logs.

The problem isn’t that validation is hard. It’s that nobody builds it into the pipeline from the start.

Level 1: Type-Safe Schemas with Pydantic

The foundation is a schema — a formal definition of what correct data looks like.

Defining the Schema

Pydantic v2 gives you runtime type checking with a familiar Python syntax:

from pydantic import BaseModel, Field
from typing import Optional

class ProductSchema(BaseModel):
    title: str
    price: float = Field(ge=0.01)       # Must be positive
    availability: bool
    rating: Optional[float] = Field(default=None, ge=0, le=5)

This schema enforces: title is a non-null string, price is a positive float, availability is a boolean, and rating (if present) is between 0 and 5.

ScraperGuard’s Schema API

ScraperGuard wraps Pydantic with a BaseSchema class and validators module that simplifies the declaration:

from scraperguard.schema import BaseSchema, validators

class ProductSchema(BaseSchema):
    title: str
    price: float = validators.range(min=0.01)
    availability: bool
    rating: float = validators.range(min=0, max=5)

The validators module provides:

validators.range(min=0.01)                    # → Field(ge=0.01)
validators.range(min=0, max=5)                # → Field(ge=0, le=5)
validators.pattern(r"^[A-Z]{2}-\d{6}$")      # → Field(pattern=...)
validators.cardinality(min=1, max=20)         # → List length constraint
validators.not_null()                          # → Explicit non-null marker

These are thin wrappers around Pydantic’s Field() function — no magic, just cleaner syntax for common constraints.

How validate_item Works Internally

ScraperGuard’s BaseSchema.validate_item() maps Pydantic validation errors to structured failure objects:

# From scraperguard/core/schema/base.py — actual implementation
@classmethod
def validate_item(cls, item: dict) -> tuple[bool, list[FieldFailure]]:
    """Validate a single item against this schema.

    Returns (passed: bool, failures: list[FieldFailure])
    """
    try:
        cls.model_validate(item)
        return True, []
    except ValidationError as e:
        failures = []
        for error in e.errors():
            field_name = str(error["loc"][0]) if error["loc"] else "unknown"
            error_type = error["type"]

            # Map Pydantic error types to ScraperGuard failure types
            if error_type == "missing":
                failure_type = "missing"
            elif error_type == "none_not_allowed" or "none" in error_type:
                failure_type = "null"
            elif "type" in error_type:
                failure_type = "type_mismatch"
            else:
                failure_type = "validation_error"

            failures.append(FieldFailure(
                field_name=field_name,
                failure_type=failure_type,
                count=1,
                details=error.get("msg", ""),
            ))
        return False, failures

The FieldFailure dataclass captures exactly what went wrong:

# From scraperguard/storage/models.py
@dataclass
class FieldFailure:
    field_name: str      # Which field failed
    failure_type: str    # "missing", "null", "type_mismatch", "validation_error"
    count: int           # How many items had this failure
    details: str = ""    # Human-readable error message

    def to_dict(self) -> dict:
        return {
            "field_name": self.field_name,
            "failure_type": self.failure_type,
            "count": self.count,
            "details": self.details,
        }

Validating Individual Items

from pydantic import ValidationError

def validate_item(item: dict, schema=ProductSchema) -> tuple[bool, list[str]]:
    try:
        schema.model_validate(item)
        return True, []
    except ValidationError as e:
        errors = []
        for error in e.errors():
            field = error["loc"][0] if error["loc"] else "unknown"
            errors.append(f"{field}: {error['type']}")
        return False, errors

Test it:

# Valid item
validate_item({"title": "Headphones", "price": 79.99, "availability": True, "rating": 4.3})
# → (True, [])

# Missing required field
validate_item({"title": "Headphones", "availability": True})
# → (False, ["price: missing"])

# Out of range
validate_item({"title": "Headphones", "price": -5.0, "availability": True, "rating": 47.3})
# → (False, ["price: greater_than_equal", "rating: less_than_equal"])

This is useful for filtering. But for monitoring, you need batch-level statistics.

Level 2: Batch Validation with Statistics

Validating individual items tells you “this item is broken.” Batch validation tells you “this scraper run is degraded.”

One broken item out of 1,000 is noise. Three hundred broken items out of 1,000 is a catastrophe. You need aggregate statistics to tell the difference.

Computing Batch Statistics

from collections import defaultdict

def validate_batch(items: list[dict], schema=ProductSchema) -> dict:
    total = len(items)
    passed_count = 0
    field_failures = defaultdict(lambda: defaultdict(int))

    field_names = list(schema.model_fields.keys())
    null_counts = {field: 0 for field in field_names}

    for item in items:
        # Count nulls/missing for every field
        for field in field_names:
            value = item.get(field)
            if value is None or field not in item:
                null_counts[field] += 1

        # Validate against schema
        passed, errors = validate_item(item, schema)
        if passed:
            passed_count += 1
        else:
            for error_str in errors:
                field, error_type = error_str.split(": ", 1)
                field_failures[field][error_type] += 1

    # Compute null ratios
    null_ratios = {
        field: null_counts[field] / total if total > 0 else 0.0
        for field in field_names
    }

    return {
        "total_items": total,
        "passed": passed_count,
        "failed": total - passed_count,
        "pass_rate": passed_count / total if total > 0 else 0.0,
        "null_ratios": null_ratios,
        "field_failures": dict(field_failures),
    }

How ScraperGuard Does It

ScraperGuard’s BaseSchema.validate_batch() computes the same statistics in a single call:

result = ProductSchema.validate_batch(items, run_id="run-001", url="https://shop.example.com")

# result is a ValidationResult with:
#   result.total_items    → 30
#   result.passed_count   → 24
#   result.failed_count   → 6
#   result.null_ratios    → {"title": 0.0, "price": 0.2, "availability": 0.0, "rating": 0.1}
#   result.field_failures → [FieldFailure(field="price", error="missing", count=6)]

The ValidationResult is a structured object that feeds into the health score computation and drift detection. Null ratios are the critical metric — they’re what enable drift detection.

The ValidationResult is the core data object that feeds everything downstream — drift detection, health scoring, and reporting:

# From scraperguard/storage/models.py
@dataclass
class ValidationResult:
    id: str                              # Unique result ID
    run_id: str                          # Which run this belongs to
    url: str                             # URL being validated
    schema_name: str                     # Schema class name
    total_items: int                     # Total items in batch
    passed_count: int                    # Items that passed validation
    failed_count: int                    # Items that failed
    field_failures: list[FieldFailure]   # Per-field failure details
    null_ratios: dict[str, float]        # Per-field null ratio (0.0-1.0)
    timestamp: datetime                  # When validation ran

Example with real numbers:

result = ProductSchema.validate_batch(items, run_id="run-001", url="https://shop.example.com")

# Access the results:
print(f"Pass rate: {result.passed_count / result.total_items:.0%}")  # 70%
print(f"Price null ratio: {result.null_ratios['price']:.0%}")         # 10%
print(f"Rating null ratio: {result.null_ratios['rating']:.0%}")       # 10%

# Field failures broken down:
for ff in result.field_failures:
    print(f"  {ff.field_name}: {ff.failure_type} ({ff.count} items)")
# Output:
#   price: missing (1 items)
#   price: validation_error (1 items)    <- negative price
#   rating: validation_error (1 items)   <- rating > 5

Level 3: Null Drift Detection

This is where validation gets interesting. Instead of just checking “is this run valid?”, you check “is this run different from what’s normal?”

The Concept

Drift detection works by comparing the current run’s null ratios against a historical baseline:

After every run, save the null ratios to storage.
Before analyzing a new run, load null ratios from the last N runs to compute a baseline average.
Compare the current run’s ratios against the baseline.
If any field shifted significantly, flag it with a severity level.

Why Baselines Matter

Consider this: your scraper reports 15% null prices. Is that a problem?

If the baseline is 12%, a 15% ratio is noise. Not worth investigating.
If the baseline is 1%, a 15% ratio is a 14 percentage point jump — something definitely changed.
If the baseline is 40%, a 15% ratio is actually an improvement.

Without baselines, you can’t distinguish normal variation from real problems. Every threshold you set will either generate false alarms (too sensitive) or miss real issues (too loose).

Implementation

import json
import sqlite3
from datetime import datetime

class DriftDetector:
    def __init__(self, db_path: str = "scraper_metrics.db"):
        self.conn = sqlite3.connect(db_path)
        self._create_tables()

    def _create_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS validation_history (
                id INTEGER PRIMARY KEY,
                url TEXT NOT NULL,
                schema_name TEXT NOT NULL,
                null_ratios TEXT NOT NULL,
                total_items INTEGER,
                pass_rate REAL,
                timestamp TEXT NOT NULL
            )
        """)
        self.conn.commit()

    def get_baseline(self, url: str, schema_name: str, lookback: int = 5) -> dict:
        cursor = self.conn.execute(
            """SELECT null_ratios FROM validation_history
               WHERE url = ? AND schema_name = ?
               ORDER BY timestamp DESC LIMIT ?""",
            (url, schema_name, lookback)
        )
        rows = cursor.fetchall()
        if not rows:
            return {}

        all_ratios = [json.loads(row[0]) for row in rows]
        fields = set()
        for r in all_ratios:
            fields.update(r.keys())

        return {
            field: sum(r.get(field, 0.0) for r in all_ratios) / len(all_ratios)
            for field in fields
        }

    def detect_drift(self, current: dict, baseline: dict) -> list[dict]:
        events = []
        for field, current_ratio in current.items():
            if field not in baseline:
                continue

            previous = baseline[field]
            delta = current_ratio - previous

            if abs(delta) <= 0.15:  # Below threshold
                continue

            if abs(delta) > 0.5:
                severity = "critical"
            elif abs(delta) > 0.3:
                severity = "warning"
            else:
                severity = "info"

            direction = "increased" if delta > 0 else "decreased"
            events.append({
                "field": field,
                "current": current_ratio,
                "baseline": previous,
                "delta": delta,
                "severity": severity,
                "message": f"'{field}' null ratio {direction} from {previous:.1%} to {current_ratio:.1%} ({delta:+.1%})"
            })

        events.sort(key=lambda e: abs(e["delta"]), reverse=True)
        return events

    def save_result(self, url: str, schema_name: str, result: dict):
        self.conn.execute(
            "INSERT INTO validation_history VALUES (NULL, ?, ?, ?, ?, ?, ?)",
            (url, schema_name, json.dumps(result["null_ratios"]),
             result["total_items"], result["pass_rate"], datetime.utcnow().isoformat())
        )
        self.conn.commit()

Critical Implementation Detail: Save Order

There’s a subtle bug that many implementations get wrong. If you save the current validation result before computing the baseline, the current run is included in its own baseline. This dilutes the detected delta:

# ❌ WRONG: save before drift detection
storage.save_validation_result(current_result)       # Current is now in storage
baseline = detector.get_baseline(url, schema_name)   # Baseline includes current!
events = detector.detect_drift(current_ratios, baseline)  # Delta is diluted

# ✓ CORRECT: detect drift first, then save
baseline = detector.get_baseline(url, schema_name)   # Baseline is historical only
events = detector.detect_drift(current_ratios, baseline)  # Delta is accurate
storage.save_validation_result(current_result)       # Now save for future baselines

This was an actual bug I fixed in ScraperGuard during development. The baseline of 3.3% + 80.0% averaged to 41.7%, producing a delta of 38.3% (warning) instead of the correct 76.7% (critical). The order of operations matters.

The drift detector produces structured DriftEvent objects:

# From scraperguard/core/schema/drift.py
@dataclass
class DriftEvent:
    field_name: str        # Which field drifted
    current_ratio: float   # Current null ratio (0.0-1.0)
    baseline_ratio: float  # Historical average
    delta: float           # current - baseline
    severity: str          # "critical", "warning", "info"
    message: str           # Human-readable description

# Severity thresholds (from the actual implementation):
# |delta| > 0.50 -> "critical"  (likely total breakage)
# |delta| > 0.30 -> "warning"   (significant degradation)
# |delta| > 0.15 -> "info"      (notable, worth watching)
# |delta| <= 0.15 -> ignored     (normal variation)

The run_drift_analysis() function handles the end-to-end flow:

from scraperguard.core.schema.drift import run_drift_analysis

# This retrieves historical results from storage, computes baseline,
# and returns drift events — all in one call.
drift_events = run_drift_analysis(
    current_result=result,    # Current ValidationResult
    storage=storage,          # StorageBackend instance
    baseline_count=5,         # How many historical runs to average
    threshold=0.15,           # Minimum delta to report
)

# Returns empty list if no historical data (first run)
# Returns sorted events (highest |delta| first) when drift detected

Putting It Together

# After every scraper run:
url = "https://shop.example.com/products"
items = my_scraper.run(url)
result = validate_batch(items)

# Drift detection
baseline = detector.get_baseline(url, "ProductSchema")
if baseline:
    drift_events = detector.detect_drift(result["null_ratios"], baseline)
    for event in drift_events:
        if event["severity"] == "critical":
            send_slack_alert(event["message"])
        print(f"[{event['severity']}] {event['message']}")

# Save AFTER drift detection
detector.save_result(url, "ProductSchema", result)

Normal run output: nothing (no drift detected).

Bad run output (selector broke):

[critical] 'price' null ratio increased from 3.2% to 80.0% (+76.8%)
[warning] 'availability' null ratio increased from 1.5% to 45.0% (+43.5%)

Level 4: The Complete Validation Pipeline

Here’s everything assembled into a reusable pipeline:

class ScraperValidator:
    def __init__(self, schema, db_path="scraper_metrics.db"):
        self.schema = schema
        self.detector = DriftDetector(db_path)

    def validate_run(self, url: str, items: list[dict]) -> dict:
        # Step 1: Batch validation
        result = validate_batch(items, self.schema)

        # Step 2: Drift detection (before saving!)
        baseline = self.detector.get_baseline(url, self.schema.__name__)
        drift_events = self.detector.detect_drift(result["null_ratios"], baseline) if baseline else []

        # Step 3: Save for future baselines
        self.detector.save_result(url, self.schema.__name__, result)

        # Step 4: Health assessment
        score = result["pass_rate"] * 100
        has_critical = any(e["severity"] == "critical" for e in drift_events)
        if has_critical:
            score = min(score, 30)

        return {
            "url": url,
            "validation": result,
            "drift_events": drift_events,
            "health_score": round(score),
            "status": "healthy" if score >= 80 else "degraded" if score >= 50 else "critical",
        }

Usage:

validator = ScraperValidator(ProductSchema)
report = validator.validate_run("https://shop.example.com", items)
print(f"Health: {report['health_score']}/100 ({report['status']})")

How ScraperGuard Extends This

The validation layer we built above handles the data quality side. ScraperGuard extends it with:

Structural DOM diffing — Compares normalized HTML trees to detect what changed in the page structure, not just the data. Catches problems before selectors break.
Selector tracking — Monitors CSS selector match counts between runs. Immediate alert when a selector stops matching.
Failure classification — Connects data quality failures to root causes: SELECTOR_BREAK, CAPTCHA_INJECTION, DOM_RESTRUCTURE, RATE_LIMIT, AB_VARIANT, etc.
4-component health scoring — Weights schema compliance (30%), extraction completeness (30%), selector stability (25%), and structural stability (15%) into a single 0-100 score.
Persistent storage — SQLite backend stores all snapshots, validation results, and run metadata.
Alerting — Slack, email, and webhook dispatchers fire when thresholds are breached.

The Python API handles everything in one call:

result = ProductSchema.validate_batch(items, run_id="run-001", url="https://shop.example.com")

And the CLI runs the complete pipeline:

scraperguard run https://shop.example.com --schema product_schema.py --selectors ".price,.title"

The validation and drift detection logic in ScraperGuard follows the exact same principles we built from scratch in this tutorial — the same baseline computation, the same severity thresholds, the same save-order requirement.

Complete Pipeline Example

Here’s the full pipeline running end-to-end — from defining a schema to getting a health report:

from scraperguard.storage.sqlite import SQLiteBackend
from scraperguard.core.snapshot.capture import capture_snapshot
from scraperguard.core.schema.drift import run_drift_analysis
from scraperguard.core.dom_diff.parser import parse_to_tree
from scraperguard.core.dom_diff.selector_tracker import track_selectors
from scraperguard.core.classify.classifier import ClassificationInput, classify_failure
from scraperguard.health import compute_health_score, format_health_report
from scraperguard.storage.models import SnapshotMetadata
from datetime import datetime, timezone

# Setup
storage = SQLiteBackend("scraperguard.db")
run = storage.create_run("product_scraper")

# 1. Validate
result = ProductSchema.validate_batch(items, run_id=run.id, url=url)

# 2. Detect drift (BEFORE saving — critical ordering!)
drift_events = run_drift_analysis(result, storage, baseline_count=5)

# 3. Save for future baselines
storage.save_validation_result(result)

# 4. Capture and diff snapshots
metadata = SnapshotMetadata(
    http_status=200, latency_ms=150.0,
    timestamp=datetime.now(timezone.utc),
    headers={"content-type": "text/html"},
    response_size_bytes=len(html.encode()),
)
snapshot = capture_snapshot(url, html, items, metadata, storage, run_id=run.id)

# 5. Track selectors
tree = parse_to_tree(snapshot.normalized_html)
prev_snapshot = storage.get_latest_snapshot(url)
prev_tree = parse_to_tree(prev_snapshot.normalized_html) if prev_snapshot else None
selector_statuses = track_selectors(tree, prev_tree, [".price", ".title", ".stock"])

# 6. Classify failures
classifications = classify_failure(ClassificationInput(
    validation_result=result,
    dom_changes=[],
    selector_statuses=selector_statuses,
    raw_html=html,
    http_status=200,
    response_size_bytes=len(html.encode()),
))

# 7. Health score
report = compute_health_score(
    validation_result=result,
    selector_statuses=selector_statuses,
    dom_changes=[],
    classifications=classifications,
    drift_events=drift_events,
    run_id=run.id, url=url,
)

print(format_health_report(report))

Quick Reference

Level	What It Does	What It Catches	Effort
Schema Validation	Type + constraint checks per item	Wrong types, missing fields, out-of-range values	Low
Batch Statistics	Aggregate pass rates + null ratios	Overall run degradation, field-level issues	Low
Null Drift Detection	Current vs. historical baseline comparison	Gradual degradation, sudden breakage, selector drift	Medium
Full Pipeline	All above + health scoring + alerting	Everything, with specific diagnostics	ScraperGuard handles this

Start with Level 1 (schema validation) — it takes 15 minutes to add and catches the majority of silent failures. Add drift detection when you need historical context. Or use ScraperGuard to get the full pipeline out of the box.

On This Page

Building Schema Validation for Scraping Pipelines: From Type Checks to Drift Detection