Production Web Scraping Pipelines in Python

TL;DR

• A production pipeline separates scheduling, bounded concurrency, failure recovery, extraction, validation, and observability so each layer can fail without corrupting downstream data.
• Stateful scheduling controls when each source runs, while domain-specific limits prevent workers from overwhelming target sites or local resources.
• Retry policies classify failures before applying backoff, proxy rotation, quarantine, or permanent rejection.
• Layered extraction prefers structured page data, records confidence for selector fallbacks, and validates normalized output before an AI pipeline receives it.
• Custom infrastructure makes sense when browser control or scale offsets its maintenance cost. Managed APIs cost less when you mainly need consistent Markdown or JSON across diverse sites.

Architecture at a glance: the build-versus-managed decision

Your choice depends on how much control the workload requires and how much crawler maintenance you can support. High request volume alone does not settle the question. Site diversity, session state, anti-bot changes, and adapter churn often consume more engineering time than raw fetching.

Decision factor	Build custom	Use a managed API like Context.dev
Crawl volume and site diversity	Prefer custom infrastructure for sustained high volume across a limited set of stable sites.	Prefer managed extraction when you cover many changing sites and per-site maintenance dominates cost.
Maintenance capacity	Build when dedicated engineers can own schedulers, proxies, browsers, adapters, and incident response.	Use an API when crawler maintenance would compete with product or data work.
Browser requirements	Use Playwright when workflows need persistent sessions, complex interactions, or precise browser control.	Use managed retrieval for one-off page extraction, crawling, and rendered content without long-lived sessions.
Anti-bot burden	Build when you need direct control over fingerprints, proxy selection, and request behavior.	Use an API when you do not want to maintain proxy rotation, browser infrastructure, and blocking responses.

At Context.dev, we provide managed retrieval, rendering, and extraction for developers who need clean Markdown or schema-shaped JSON. The service can replace internal scheduling, retry, proxy, browser, and site-adapter infrastructure when the goal is consistent input for AI or LLM consumers.

Custom Playwright automation remains the better option when browser state and interaction details define the task. For lower-level tool selection, consult the existing Python library roundup. The resilient scraper architecture guide covers deeper recovery patterns, while the sections below focus on connecting those decisions into one production pipeline.

Scheduling and orchestrating crawls at scale

A production crawler needs a durable frontier that records what to fetch, when to fetch it, and why it has priority. Each frontier row should include the URL, domain, priority tier, next crawl time, fetch state, and cache validators such as ETag and Last-Modified. Store that state in PostgreSQL, SQLite, or another transactional database rather than relying on scheduler memory.

The following APScheduler pattern dispatches due URLs into a priority queue while enforcing a budget for each domain. Lower priority numbers represent more urgent work.

import queue
import sqlite3
import time
from collections import defaultdict
from urllib.parse import urlparse
 
from apscheduler.schedulers.background import BackgroundScheduler
 
DB = "frontier.db"
DOMAIN_BUDGET = 20
work_queue = queue.PriorityQueue(maxsize=2_000)
 
def connect():
    return sqlite3.connect(DB)
 
def initialize():
    with connect() as db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS frontier (
                url TEXT PRIMARY KEY,
                domain TEXT NOT NULL,
                priority INTEGER NOT NULL DEFAULT 100,
                next_fetch REAL NOT NULL,
                state TEXT NOT NULL DEFAULT 'pending',
                etag TEXT,
                last_modified TEXT
            )
        """)
 
def dispatch_due():
    now = time.time()
    counts = defaultdict(int)
 
    with connect() as db:
        rows = db.execute("""
            SELECT url, domain, priority
            FROM frontier
            WHERE state = 'pending' AND next_fetch <= ?
            ORDER BY priority ASC, next_fetch ASC
            LIMIT 1000
        """, (now,)).fetchall()
 
        for url, domain, priority in rows:
            if counts[domain] >= DOMAIN_BUDGET:
                continue
 
            try:
                work_queue.put_nowait((priority, now, url))
            except queue.Full:
                break
 
            db.execute(
                "UPDATE frontier SET state = 'queued' WHERE url = ?",
                (url,),
            )
            counts[domain] += 1
 
initialize()
 
scheduler = BackgroundScheduler()
scheduler.add_job(dispatch_due, "interval", seconds=10, max_instances=1)
scheduler.start()

A database state change prevents the next scheduler tick from enqueuing the same URL again. After a worker finishes, it should update the row with a new next_fetch time and return the state to pending. A failed fetch can instead enter a delayed retry state. Larger deployments can replace the in-process queue with Redis, RabbitMQ, or a cloud queue. Celery beat can run the periodic dispatcher, while Celery workers consume the resulting jobs.

Incremental crawling reduces bandwidth and extraction work. Compare each discovered sitemap against the previous snapshot, and enqueue only new URLs or entries whose modification timestamps changed. For ordinary pages, send stored ETag and Last-Modified values through If-None-Match and If-Modified-Since headers. A 304 Not Modified response lets the worker postpone extraction and schedule the next check.

Re-crawl cadence should reflect source behavior. You might check a news homepage every few minutes, an article daily, and an archived document monthly. Track observed change intervals and lengthen the cadence when repeated checks find no changes. Shorten it when a source changes more frequently. Context.dev Monitors can manage scheduled crawling and change detection when maintaining frontier state, snapshots, and diff logic no longer supports your core product.

Bounded concurrency without overwhelming targets or your own infrastructure

Bounded concurrency lets a crawler use spare capacity across domains without sending an unsafe burst to any one target. A global limit protects your sockets, memory, browser processes, and proxy capacity. A separate gate for each domain controls simultaneous requests and spaces request starts.

import asyncio
from collections import defaultdict
from contextlib import asynccontextmanager
from urllib.parse import urlparse
 
import httpx
 
GLOBAL_CONCURRENCY = 40
global_limit = asyncio.Semaphore(GLOBAL_CONCURRENCY)
 
fetch_queue = asyncio.PriorityQueue(maxsize=2_000)
extract_queue = asyncio.Queue(maxsize=200)
validate_queue = asyncio.Queue(maxsize=100)
 
class DomainGate:
    def __init__(self, concurrency=2, requests_per_second=1.0):
        self.semaphore = asyncio.Semaphore(concurrency)
        self.start_lock = asyncio.Lock()
        self.interval = 1.0 / requests_per_second
        self.next_start = 0.0
 
    @asynccontextmanager
    async def enter(self):
        await self.semaphore.acquire()
        try:
            async with self.start_lock:
                loop = asyncio.get_running_loop()
                delay = self.next_start - loop.time()
                if delay > 0:
                    await asyncio.sleep(delay)
                self.next_start = loop.time() + self.interval
            yield
        finally:
            self.semaphore.release()
 
domain_gates = defaultdict(
    lambda: DomainGate(concurrency=2, requests_per_second=1.0)
)
 
async def fetch_worker(client):
    while True:
        priority, sequence, url = await fetch_queue.get()
        domain = urlparse(url).netloc
 
        try:
            async with global_limit:
                async with domain_gates[domain].enter():
                    response = await client.get(url, timeout=20)
 
            await extract_queue.put(
                (url, response.status_code, response.text)
            )
        finally:
            fetch_queue.task_done()
 
async def extraction_worker(extract):
    while True:
        url, status, html = await extract_queue.get()
        try:
            record = await asyncio.to_thread(extract, url, status, html)
            await validate_queue.put(record)
        finally:
            extract_queue.task_done()

The domain gate controls both in-flight requests and request-start frequency. A semaphore alone limits concurrency but can still produce bursts when several fast requests finish together. The start lock spaces requests according to the configured rate.

Set global concurrency no higher than the smallest relevant capacity limit. Those limits include safe proxy concurrency, available connections, browser memory, and the combined tolerance of active domains. If a proxy provider safely supports 100 concurrent requests but local browser workers support 20, a limit of 20 protects the actual bottleneck. Per-domain settings should begin conservatively and adjust in response to latency, 429 responses, and published crawl guidance.

Bounded queues provide backpressure between stages. When extraction falls behind, extract_queue.put() blocks fetch workers once the queue reaches 200 items. Fetch throughput then declines instead of consuming more memory. The same mechanism slows extraction when validation fills its queue. Monitor queue depth and item age because a consistently full queue identifies an undersized downstream stage, while an empty fetch queue usually points to scheduling or discovery limits.

Retry policies, backoff, and jitter that don't make things worse

Retry logic should classify a failure before deciding whether another request can help. A blanket retry loop wastes capacity on missing pages and malformed responses while giving rate limits too few recovery options.

from dataclasses import dataclass
from enum import Enum
import asyncio
import random
 
class Action(Enum):
    RETRY = "retry"
    QUARANTINE = "quarantine"
    DROP = "drop"
    BLOCK = "block"
 
@dataclass(frozen=True)
class RetryDecision:
    action: Action
    max_attempts: int = 1
    delay_seconds: float = 0
 
def full_jitter(attempt: int, base: float = 1, cap: float = 60) -> float:
    ceiling = min(cap, base * (2 ** attempt))
    return random.uniform(0, ceiling)
 
def retry_decision(
    *,
    attempt: int,
    status: int | None = None,
    timed_out: bool = False,
    malformed: bool = False,
    permanent_block: bool = False,
) -> RetryDecision:
    if permanent_block:
        return RetryDecision(Action.BLOCK)
 
    if status == 404:
        return RetryDecision(Action.DROP)
 
    if malformed:
        return RetryDecision(Action.QUARANTINE)
 
    if status == 429:
        limit = 6
        action = Action.RETRY if attempt < limit else Action.BLOCK
        return RetryDecision(action, limit, full_jitter(attempt, 5, 300))
 
    if timed_out or status in {502, 503, 504}:
        limit = 4
        action = Action.RETRY if attempt < limit else Action.QUARANTINE
        return RetryDecision(action, limit, full_jitter(attempt, 1, 60))
 
    return RetryDecision(Action.DROP)
 
async def fetch_with_retries(fetch):
    attempt = 0
 
    while True:
        result = await fetch()
        decision = retry_decision(
            attempt=attempt,
            status=result.status,
            timed_out=result.timed_out,
            malformed=result.malformed,
            permanent_block=result.permanent_block,
        )
 
        if decision.action is not Action.RETRY:
            return result, decision
 
        await asyncio.sleep(decision.delay_seconds)
        attempt += 1

Full jitter chooses a random delay between zero and the current exponential ceiling. Randomization prevents hundreds of workers from retrying together after a shared outage. A production implementation should also honor a valid Retry-After response header, using it as the minimum wait for a 429.

Each domain also needs its own circuit breaker. Track failures in a rolling window, and open the breaker when a domain crosses a threshold such as 20 failures among its last 25 requests. An open breaker should stop dispatching work for that domain while other domains continue. After a cooldown, send one probe request through a half-open breaker. Close it after a successful probe, or reopen it after another failure.

Circuit breakers should count failures that indicate domain-wide trouble, including repeated 503 responses and blocks across several healthy proxies. They should not open because one proxy timed out. Keeping domain health separate from proxy health prevents one bad endpoint from consuming the worker pool or exhausting every available proxy.

Proxy rotation and anti-bot handling as a resilience layer

Proxy selection should execute the action chosen by the retry layer. For example, a timeout can retry through the same healthy proxy, while a 429 can pause the domain and select a different identity. A detected permanent block should open the domain circuit instead of rotating through the entire pool.

Sticky sessions work well when a site issues cookies, assigns regional content, or checks whether successive requests keep the same network identity. Bind one proxy and cookie jar to a domain for a limited session window. Rotate after repeated throttling, declining proxy health, or session expiry rather than after every request.

Datacenter proxies usually offer lower cost and latency, so they suit tolerant sites and high-volume public pages. Residential proxies can help when a target treats hosting-provider addresses more aggressively, but their higher price requires stricter request budgets. Start each domain with the least expensive pool that meets its observed failure rate, then escalate only after the retry policy identifies a likely network block.

A proxy manager should score individual endpoints instead of treating a pool as interchangeable. Update the score with connection failures, latency, and block responses. Temporary failures can reduce a proxy's selection weight. Repeated failures across unrelated domains should retire it for a cooldown period. Domain-specific blocks should prevent that proxy and domain pairing without discarding an otherwise healthy endpoint.

Block detection needs page-level evidence because a successful HTTP status does not guarantee useful content. Soft blocks often return a CAPTCHA, a consent challenge, or an empty rendered document with status 200. Compare the response against expected markers such as a product title or article body. Treat hidden links and fields as possible honeypots, and do not click or submit elements that normal users cannot see.

Hard blocks usually produce repeated 401 or 403 responses, explicit access-denied pages, or connection termination across multiple known-good identities. A single 403 should not trigger endless rotation. Repeated agreement across proxies provides stronger evidence that the domain, account, or browser profile has been blocked.

Anti-bot maintenance becomes expensive when success depends on browser fingerprint consistency and accurate TLS behavior. Headless-browser detection adds another moving target. If your output requirement is clean Markdown or schema-shaped JSON, routing retrieval and rendering through a managed API such as Context.dev can cost less than maintaining proxy pools, browser profiles, CAPTCHA services, and site-specific block detectors.

Managed retrieval does not replace direct browser control in every workflow. Keep Playwright or another browser automation layer when you need persistent authenticated sessions, complex interactions, or exact control over browser state. Use a managed extraction API when web data is the product input and maintaining anti-bot infrastructure does not create a useful advantage.

Layered extraction for inconsistent HTML

A reliable extractor gives each field several ordered sources instead of giving each site one brittle selector. Embedded data usually carries clearer semantics than rendered text, while DOM heuristics remain useful when a page exposes no stable structure.

Use the following precedence for every field.

1. Read JSON-LD first, followed by microdata and meta tags.
2. Try site-specific selectors in priority order.
3. Apply DOM-shape or text heuristics as a last resort.

Each successful extraction should retain its source layer and a confidence score. The score lets validation treat a JSON-LD price differently from a currency-shaped string found somewhere in the page.

import json
import re
from dataclasses import dataclass
from typing import Callable
 
from bs4 import BeautifulSoup
 
@dataclass(frozen=True)
class FieldValue:
    value: str
    source: str
    confidence: float
 
def text_of(node) -> str | None:
    if not node:
        return None
    value = node.get("content") or node.get_text(" ", strip=True)
    return value.strip() if value and value.strip() else None
 
def from_json_ld(soup: BeautifulSoup, field: str) -> str | None:
    for script in soup.select('script[type="application/ld+json"]'):
        try:
            payload = json.loads(script.string or "")
        except json.JSONDecodeError:
            continue
 
        nodes = payload.get("@graph", []) if isinstance(payload, dict) else payload
        if isinstance(nodes, dict):
            nodes = [nodes]
        if not isinstance(nodes, list):
            nodes = [payload]
 
        for node in nodes:
            if not isinstance(node, dict):
                continue
            types = node.get("@type", [])
            types = [types] if isinstance(types, str) else types
            if "Product" not in types:
                continue
 
            if field == "title":
                return node.get("name")
 
            if field == "price":
                offers = node.get("offers", {})
                offers = offers[0] if isinstance(offers, list) and offers else offers
                if isinstance(offers, dict) and offers.get("price") is not None:
                    return str(offers["price"])
    return None
 
def from_microdata(soup: BeautifulSoup, field: str) -> str | None:
    itemprop = {"title": "name", "price": "price"}[field]
    return text_of(soup.select_one(f'[itemprop="{itemprop}"]'))
 
def from_meta(soup: BeautifulSoup, field: str) -> str | None:
    selector = {
        "title": 'meta[property="og:title"]',
        "price": 'meta[property="product:price:amount"]',
    }[field]
    return text_of(soup.select_one(selector))
 
def from_selectors(
    soup: BeautifulSoup, selectors: list[str]
) -> str | None:
    for selector in selectors:
        value = text_of(soup.select_one(selector))
        if value:
            return value
    return None
 
def from_heuristics(soup: BeautifulSoup, field: str) -> str | None:
    if field == "title":
        return text_of(soup.find(["h1", "h2"]))
 
    pattern = re.compile(r"[$€£]\s?\d+(?:[.,]\d{2})?")
    match = soup.find(string=pattern)
    return match.strip() if match else None
 
def extract_field(
    soup: BeautifulSoup,
    field: str,
    selectors: list[str],
) -> FieldValue | None:
    layers: list[tuple[str, float, Callable[[], str | None]]] = [
        ("json-ld", 0.98, lambda: from_json_ld(soup, field)),
        ("microdata", 0.90, lambda: from_microdata(soup, field)),
        ("meta", 0.82, lambda: from_meta(soup, field)),
        ("selector", 0.70, lambda: from_selectors(soup, selectors)),
        ("heuristic", 0.40, lambda: from_heuristics(soup, field)),
    ]
 
    for source, confidence, extractor in layers:
        value = extractor()
        if value:
            return FieldValue(value, source, confidence)
    return None
 
html = """
 
"""
soup = BeautifulSoup(html, "html.parser")
 
record = {
    "title": extract_field(soup, "title", ["h1.product-title", "h1"]),
    "price": extract_field(soup, "price", [".price", "[data-price]"]),
}
print(record)

Confidence values encode extraction policy rather than statistical certainty. You should tune them with observed errors for each domain. A downstream validator might accept a heuristic title for review while rejecting a heuristic price before the record reaches an AI application.

Separating raw extraction from normalization

Raw captures and canonical records should remain separate because parsing rules change more often than source pages do. Store the exact response before extraction, then run normalization as a pure transformation that never modifies the capture.

The capture should include the response body and enough metadata to reproduce the parsing decision. A content-addressed file store provides a simple immutable implementation.

import gzip
import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from pathlib import Path
 
@dataclass(frozen=True)
class RawCapture:
    url: str
    fetched_at: str
    status: int
    content_type: str
    body: bytes
 
def save_capture(capture: RawCapture, root: Path) -> str:
    identity = (
        capture.url.encode()
        + capture.fetched_at.encode()
        + capture.body
    )
    capture_id = hashlib.sha256(identity).hexdigest()
    root.mkdir(parents=True, exist_ok=True)
 
    with gzip.open(root / f"{capture_id}.body.gz", "xb") as body_file:
        body_file.write(capture.body)
 
    metadata = {
        "id": capture_id,
        "url": capture.url,
        "fetched_at": capture.fetched_at,
        "status": capture.status,
        "content_type": capture.content_type,
    }
    (root / f"{capture_id}.json").write_text(
        json.dumps(metadata, sort_keys=True)
    )
    return capture_id
 
def parse_weight_grams(value: str | None) -> int | None:
    if not value:
        return None
    match = re.fullmatch(r"\s*([\d.]+)\s*(kg|g)\s*", value.lower())
    if not match:
        return None
    amount, unit = match.groups()
    multiplier = Decimal("1000") if unit == "kg" else Decimal("1")
    return int(Decimal(amount) * multiplier)
 
def normalize(raw_fields: dict, brand_aliases: dict[str, str]) -> dict:
    price_text = raw_fields["price"]["value"]
    price = Decimal(re.sub(r"[^\d.]", "", price_text))
 
    published = raw_fields.get("published_at")
    published_at = (
        datetime.fromisoformat(published["value"].replace("Z", "+00:00"))
        if published
        else None
    )
 
    raw_brand = raw_fields.get("brand", {}).get("value")
    canonical_brand = brand_aliases.get(raw_brand, raw_brand)
 
    return {
        "title": raw_fields["title"]["value"].strip(),
        "price": price,
        "currency": raw_fields.get("currency", {}).get("value", "USD"),
        "brand": canonical_brand,
        "weight_g": parse_weight_grams(
            raw_fields.get("weight", {}).get("value")
        ),
        "published_at": published_at,
        "evidence": {
            name: {
                "source": field["source"],
                "confidence": field["confidence"],
            }
            for name, field in raw_fields.items()
            if isinstance(field, dict) and "source" in field
        },
    }

A parser can read either captured HTML or captured JSON and produce raw_fields. Updated normalization code can then reprocess every saved capture without another request. Retaining the original response also lets you distinguish a broken parser from a changed source page during debugging.

Schema changes become cheaper under the same design. You can add a normalized field, revise an entity alias, or correct a unit conversion by replaying stored captures instead of recrawling every URL.

Validating canonical output with Pydantic before data reaches an AI pipeline

Pydantic should enforce the final contract before any record enters an embedding job, prompt, agent tool, or model context. Validation should inspect both normalized values and the extraction evidence that produced them.

The following Pydantic v2 model rejects missing commercial fields and low-confidence evidence. Failed records enter quarantine with the validation details rather than disappearing or passing downstream.

import json
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import Literal
 
from pydantic import (
    BaseModel,
    Field,
    HttpUrl,
    ValidationError,
    field_validator,
    model_validator,
)
 
class Evidence(BaseModel):
    source: Literal[
        "json-ld",
        "microdata",
        "meta",
        "selector",
        "heuristic",
    ]
    confidence: float = Field(ge=0, le=1)
 
class Product(BaseModel):
    source_url: HttpUrl
    title: str = Field(min_length=2, max_length=500)
    price: Decimal | None = Field(default=None, gt=0)
    currency: str = Field(min_length=3, max_length=3)
    brand: str | None = None
    weight_g: int | None = Field(default=None, gt=0)
    published_at: datetime | None = None
    evidence: dict[str, Evidence]
 
    @field_validator("title")
    @classmethod
    def clean_title(cls, value: str) -> str:
        cleaned = " ".join(value.split())
        if not cleaned:
            raise ValueError("title is empty")
        return cleaned
 
    @field_validator("currency")
    @classmethod
    def normalize_currency(cls, value: str) -> str:
        return value.upper()
 
    @model_validator(mode="after")
    def enforce_acceptance_policy(self):
        if self.price is None:
            raise ValueError("price is required for accepted products")
 
        thresholds = {"title": 0.70, "price": 0.80}
        for field_name, minimum in thresholds.items():
            evidence = self.evidence.get(field_name)
            if evidence is None:
                raise ValueError(f"{field_name} evidence is missing")
            if evidence.confidence < minimum:
                raise ValueError(
                    f"{field_name} confidence is below {minimum}"
                )
        return self
 
def validate_or_quarantine(
    record: dict,
    quarantine_path: Path,
) -> Product | None:
    try:
        return Product.model_validate(record)
    except ValidationError as error:
        entry = {
            "record": record,
            "validation_errors": error.errors(
                include_url=False,
                include_context=False,
            ),
        }
        with quarantine_path.open("a", encoding="utf-8") as file:
            file.write(json.dumps(entry, default=str) + "\n")
        return None

The acceptance policy preserves the origin of uncertainty. A missing price and a heuristic price both fail, but quarantine records show whether extraction found nothing or found weak evidence. You can route those cases to different repair queues while allowing only validated Product objects into the AI pipeline.

Observability: logs, metrics, traces, and extraction-quality alerts

A scraper can return plausible but incorrect records without raising an exception. A site redesign might remove a price selector while every request still returns HTTP 200. Observability must therefore measure extraction quality as well as network availability.

Structured logs should record one event for each request and processing stage. Each request log should include the URL, domain, status code, latency, proxy identifier, retry count, worker identifier, and trace ID. Never log proxy credentials, session cookies, or full response bodies. Keep raw responses in access-controlled storage and include the storage key in the log instead.

Metrics should separate retrieval health from data quality. Useful retrieval metrics include request success rate, response latency, retry count, queue depth, and block rate per domain. Quality metrics should track schema-validation failures, null fields per record, and confidence distributions for important fields. Domain labels help isolate a broken adapter, but URL labels create excessive metric cardinality.

Confidence metrics expose failures that HTTP monitoring misses. Suppose a product domain normally produces title confidence near 0.95 and price confidence near 0.90. If a redesign forces the extractor onto heuristic fallbacks, both values may fall even though request success remains unchanged. Track confidence by field and source layer so you can distinguish a weak price extractor from a general page failure.

Distributed traces should connect the fetch, extract, normalize, and validate stages under one trace ID. Each stage should record its duration, result, and relevant metadata. An extraction span might record which fallback supplied each field, while a validation span might record failed field names without copying sensitive values. OpenTelemetry can send these spans to systems such as Jaeger, Grafana Tempo, or a commercial tracing service.

Quality alerts need rolling baselines because domains differ. Alert when a domain’s null-price rate exceeds its normal range, when average confidence crosses a minimum threshold, or when schema failures rise over several crawl windows. Use a short window for severe changes and a longer window for gradual drift. Uptime alerts should page operators when collection stops. Quality alerts should quarantine affected records and notify the owner before incorrect data reaches an AI pipeline.

Reference pipeline: putting the layers together

A useful reference pipeline keeps each stage replaceable while carrying one trace ID through the complete job. The following product crawler uses a scheduled dispatcher, bounded queues, per-domain concurrency, proxy-aware retries, layered extraction, normalization, validation, and structured events. Install httpx, beautifulsoup4, and pydantic before running it.

import asyncio
import json
import random
import time
import uuid
from collections import defaultdict
from contextlib import contextmanager
from decimal import Decimal
from urllib.parse import urlparse
 
import httpx
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
 
URLS = [
    "https://example.com/products/widget",
]
PROXIES = [None]
GLOBAL_LIMIT = asyncio.Semaphore(10)
DOMAIN_LIMITS = defaultdict(lambda: asyncio.Semaphore(2))
METRICS = defaultdict(int)
RAW_STORE = {}
QUARANTINE = []
 
def emit(event, **values):
    print(json.dumps({
        "event": event,
        "time": time.time(),
        **values,
    }, default=str))
 
@contextmanager
def span(name, trace_id, **values):
    started = time.perf_counter()
    try:
        yield
        emit(
            "span",
            name=name,
            trace_id=trace_id,
            duration_ms=round((time.perf_counter() - started) * 1000),
            outcome="ok",
            **values,
        )
    except Exception as exc:
        emit(
            "span",
            name=name,
            trace_id=trace_id,
            outcome="error",
            error=type(exc).__name__,
            **values,
        )
        raise
 
class RetryableBlock(Exception):
    pass
 
class Product(BaseModel):
    url: str
    title: str = Field(min_length=1)
    price: Decimal = Field(gt=0)
    confidence: float = Field(ge=0, le=1)
 
async def fetch(url, trace_id):
    domain = urlparse(url).netloc
 
    async with GLOBAL_LIMIT, DOMAIN_LIMITS[domain]:
        for attempt in range(3):
            proxy = random.choice(PROXIES)
            started = time.perf_counter()
 
            try:
                async with httpx.AsyncClient(
                    proxy=proxy,
                    timeout=15,
                    follow_redirects=True,
                ) as client:
                    response = await client.get(url)
 
                emit(
                    "request",
                    trace_id=trace_id,
                    domain=domain,
                    status=response.status_code,
                    latency_ms=round(
                        (time.perf_counter() - started) * 1000
                    ),
                    proxy=proxy,
                    retry_count=attempt,
                )
 
                if response.status_code in {403, 429, 503}:
                    raise RetryableBlock()
 
                response.raise_for_status()
 
                if "captcha" in response.text.lower():
                    raise RetryableBlock()
 
                RAW_STORE[url] = response.text
                METRICS[(domain, "fetch_ok")] += 1
                return response.text
 
            except (httpx.TimeoutException, RetryableBlock):
                METRICS[(domain, "retry")] += 1
                if attempt == 2:
                    raise
                await asyncio.sleep(random.uniform(0, 2 ** attempt))
 
def extract(html):
    soup = BeautifulSoup(html, "html.parser")
 
    for node in soup.select('script[type="application/ld+json"]'):
        try:
            value = json.loads(node.get_text())
            records = value if isinstance(value, list) else [value]
            for record in records:
                if record.get("@type") == "Product":
                    offers = record.get("offers", {})
                    return {
                        "title": (record.get("name"), 0.98),
                        "price": (offers.get("price"), 0.98),
                    }
        except (json.JSONDecodeError, AttributeError):
            pass
 
    title_meta = soup.select_one('meta[property="og:title"]')
    price_meta = soup.select_one('meta[property="product:price:amount"]')
    if title_meta and price_meta:
        return {
            "title": (title_meta.get("content"), 0.90),
            "price": (price_meta.get("content"), 0.90),
        }
 
    title = soup.select_one("h1.product-title, h1")
    price = soup.select_one("[data-price], .product-price, [class*=price]")
    return {
        "title": (title.get_text(strip=True) if title else None, 0.65),
        "price": (
            price.get("data-price") or price.get_text(strip=True)
            if price else None,
            0.55,
        ),
    }
 
def normalize(url, fields):
    raw_price = fields["price"][0] or ""
    cleaned_price = "".join(
        character for character in raw_price
        if character.isdigit() or character in ".,"
    ).replace(",", "")
 
    return {
        "url": url,
        "title": (fields["title"][0] or "").strip(),
        "price": cleaned_price,
        "confidence": min(
            fields["title"][1],
            fields["price"][1],
        ),
    }
 
async def process(url):
    trace_id = uuid.uuid4().hex
    domain = urlparse(url).netloc
 
    try:
        with span("fetch", trace_id, url=url):
            html = await fetch(url, trace_id)
 
        with span("extract", trace_id, url=url):
            fields = extract(html)
 
        with span("normalize", trace_id, url=url):
            candidate = normalize(url, fields)
 
        with span("validate", trace_id, url=url):
            product = Product.model_validate(candidate)
 
        METRICS[(domain, "validated")] += 1
        emit("accepted", trace_id=trace_id, record=product.model_dump())
 
    except Exception as exc:
        METRICS[(domain, "failed")] += 1
        QUARANTINE.append({"url": url, "error": type(exc).__name__})
        emit("quarantined", trace_id=trace_id, url=url, error=str(exc))
 
async def crawl_cycle(urls):
    queue = asyncio.Queue(maxsize=20)
 
    async def worker():
        while True:
            url = await queue.get()
            try:
                await process(url)
            finally:
                queue.task_done()
 
    workers = [asyncio.create_task(worker()) for _ in range(10)]
 
    for url in urls:
        await queue.put(url)
 
    await queue.join()
 
    for worker_task in workers:
        worker_task.cancel()
 
async def scheduler():
    while True:
        await crawl_cycle(URLS)
        await asyncio.sleep(3600)
 
asyncio.run(scheduler())

The bounded queue slows dispatch when workers fall behind, while the semaphores constrain global and per-domain traffic. Production deployments should replace the in-memory raw store, metrics, and quarantine list with durable services. The next decision layer determines whether each quarantined or failed item should be retried, escalated, or allowed to fail the crawl job.

Failure-handling decision logic in production

Production failure handling should classify the signal before choosing an action. Blanket retries waste proxy capacity on permanent failures, while immediate job failures discard recoverable work.

Failure signal	Automated response	Escalation path
A connection timeout or transient 503 occurs	Retry with full jitter and a small attempt limit. Open the domain circuit after repeated failures.	Alert when the domain circuit remains open beyond one crawl window. Fail the job only when its required-source threshold is breached.
A 429 response occurs	Respect Retry-After, lower domain concurrency, and retry through the same session when continuity matters.	Alert when throttling persists or the backlog threatens the crawl deadline.
A CAPTCHA, block page, or empty rendered DOM appears	Mark the proxy attempt as blocked, rotate according to the anti-bot policy, and retry under a lower ceiling.	Stop automated attempts when the block ceiling is reached. Quarantine the URL and alert the domain owner.
A 404 or 410 response occurs	Do not retry during the current job. Record the URL as missing and remove it from the active frontier after the configured confirmation count.	Alert only when missing pages rise sharply across the domain.
Canonical schema validation fails	Preserve the raw capture and normalized candidate, then quarantine the record.	Alert when the domain failure rate crosses its baseline. Fail the job when downstream consumers require complete coverage.
Extraction confidence falls below the acceptance threshold	Quarantine the record without refetching when the response body appears complete.	Alert on domain-level confidence drift so an adapter or fallback can be updated.
A required raw capture cannot be stored	Stop processing that record because later reprocessing would become impossible.	Fail the job when durable capture forms part of the delivery contract.

Retry decisions should depend on evidence that another attempt can change the outcome. Network timeouts, temporary throttling, and unhealthy proxies often satisfy that test. Missing selectors and invalid normalized values usually require extractor changes, so repeated fetching adds cost without improving the record.

Quarantine should preserve the URL, raw-capture key, trace ID, extractor version, confidence values, and validation errors. An operator can then replay the record after changing extraction or normalization code. The anti-bot policy should own block ceilings and proxy rotation, while the validation policy should own acceptance thresholds.

Job failure should depend on delivery requirements rather than one bad page. A crawl can finish with quarantined records when partial delivery is allowed. A required-source outage, failed raw storage, or excessive invalid-record ratio should fail the job and prevent incomplete data from reaching downstream AI consumers.

When custom infrastructure stops paying for itself

Custom infrastructure remains economical when browser behavior gives your product a specific advantage or managed request costs exceed the cost of dedicated engineering. Calculate the cost per accepted record rather than the cost per request. Include developer time, browser and proxy infrastructure, vendor fees, and incident response. Divide that total by the number of records that pass extraction-confidence and schema checks.

Maintenance costs rise as you add domains. Each site can introduce different rendering behavior, block signals, and HTML changes. Your engineers must update adapters, monitor proxy health, investigate quality drift, and revise canonical schemas. Anti-bot maintenance adds browser fingerprints, header fidelity, CAPTCHA handling, and proxy reputation management. These tasks recur even when crawl volume stays constant.

A managed API becomes attractive when your actual requirement is clean Markdown or schema-shaped JSON for an AI pipeline. Context.dev handles retrieval, rendering, retries, proxy management, and structured extraction behind one API. You still own the downstream data contract and should validate every response, but you no longer need to operate the retrieval stack or maintain site-specific extraction logic. MCP support can also connect retrieved data directly to compatible AI applications.

Compare both options with a representative crawl before committing. Measure accepted-record cost, latency, extraction coverage, and engineering hours spent on exceptions. A managed service may cost more per request while costing less per usable record because failed requests and maintenance consume less internal time.

Custom Playwright or browser automation remains the better choice when a workflow requires persistent logged-in sessions, multi-step interactions, precise browser control, or application-specific state. Managed extraction APIs fit one-off and scheduled data collection more naturally. Context.dev should not replace browser control when your application depends on the browser session itself.

Context.dev’s Python library roundup provides more detail on Scrapy, Playwright, BeautifulSoup, Parsel, and related tool choices. Its resilient scraper architecture guide covers deeper recovery patterns. Those decisions support the cost model, but operational ownership should determine whether you keep the whole pipeline in-house.

FAQs

How many concurrent workers are too many?

Worker count becomes excessive when latency, 429 responses, block rates, proxy failures, or downstream queue depth rise as concurrency increases. Set separate global and per-domain limits. Increase them gradually, and stop when accepted-record throughput flattens. Available CPU rarely provides the correct limit because target tolerance and proxy capacity usually constrain the fetch stage first.

How should I control rotating proxy costs at scale?

Route inexpensive requests through datacenter proxies and reserve residential or browser-backed retrieval for domains that require it. Track cost per accepted record by domain and retrieval mode. Sticky sessions can reduce repeated challenges, while proxy health scores prevent retries from consuming spend on failing endpoints.

How can I detect silent extraction failures?

Monitor null-field rates, confidence distributions, record counts, and schema-validation failures for each domain. Compare those measurements with recent baselines. A sudden drop in title confidence can reveal a selector break even when every request returns HTTP 200. Store raw responses so you can reproduce the failure without fetching the page again.

When should I use Scrapy instead of asyncio with httpx?

Scrapy suits broad crawls that need queues, request middleware, throttling, and crawl-state support in one framework. Asyncio with httpx suits smaller services where you want explicit control and already operate an asynchronous Python stack. Playwright belongs in either design when pages require rendering or browser interactions. Context.dev’s Python library roundup examines these tool choices in greater depth.

How does a custom pipeline differ from a managed extraction API?

A custom pipeline gives you direct control over scheduling, sessions, proxies, browsers, extraction logic, and telemetry. A managed API owns much of that retrieval and extraction work, then returns Markdown or structured JSON. You still need application-level validation, monitoring, and storage. Context.dev’s resilient scraper architecture guide covers the controls that remain important at that boundary.

Conclusion

A production scraping pipeline should optimize for trustworthy records reaching the AI or LLM layer. Successful HTTP responses have little value when extraction drift, normalization errors, or weak validation corrupt the resulting data.

Keep the pipeline in-house when persistent sessions, complex interactions, or precise browser behavior define the workload. Evaluate Context.dev when your application mainly needs current, schema-shaped web data and proxy, browser, retry, and adapter maintenance consume engineering time better spent on the product.