DEV Community: Alex Chen

Responsible Web Scraping: Rate Limiting, robots.txt, and Keeping Your Scraper Legal

Alex Chen — Mon, 23 Mar 2026 01:37:37 +0000

You can build a scraper that's technically brilliant — stealth patches, CAPTCHA solving, proxy rotation. But if you ignore rate limits, blast servers with requests, and violate terms of service, you're not a developer. You're a problem.

Let's talk about responsible scraping: how to get the data you need without being a bad actor.

Why This Matters

Beyond ethics, there are practical reasons:

Legal risk — lawsuits are real (hiQ v. LinkedIn, Clearview AI)
IP bans — aggressive scraping gets you permanently blocked
Server harm — you can accidentally DDoS small sites
Reputation — your company's IP range gets blacklisted
Data quality — rushed scraping produces worse data

Understanding robots.txt

Every website can publish a robots.txt\ file that specifies which paths scrapers should avoid:

# https://example.com/robots.txt
User-agent: *
Disallow: /admin/
Disallow: /api/internal/
Disallow: /user/*/settings
Crawl-delay: 10

User-agent: Googlebot
Allow: /
Crawl-delay: 1

Parsing robots.txt in Python

from urllib.robotparser import RobotFileParser
from urllib.parse import urlparse

class RobotsChecker:
    def __init__(self):
        self._parsers: dict[str, RobotFileParser] = {}

    def can_fetch(self, url: str, user_agent: str = "*") -> bool:
        """Check if a URL is allowed by robots.txt."""
        domain = urlparse(url).netloc

        if domain not in self._parsers:
            parser = RobotFileParser()
            parser.set_url(f"https://{domain}/robots.txt")
            try:
                parser.read()
            except Exception:
                # If we can't read robots.txt, be cautious
                return True
            self._parsers[domain] = parser

        return self._parsers[domain].can_fetch(user_agent, url)

    def get_crawl_delay(
        self, domain: str, user_agent: str = "*"
    ) -> float | None:
        """Get the recommended crawl delay."""
        if domain not in self._parsers:
            self.can_fetch(f"https://{domain}/", user_agent)

        parser = self._parsers.get(domain)
        if parser:
            delay = parser.crawl_delay(user_agent)
            return float(delay) if delay else None
        return None

# Usage
checker = RobotsChecker()

urls = [
    "https://example.com/products/123",
    "https://example.com/admin/users",
    "https://example.com/api/internal/debug",
]

for url in urls:
    allowed = checker.can_fetch(url)
    print(f"{'✓' if allowed else '✗'} {url}")

Output:

✓ https://example.com/products/123
✗ https://example.com/admin/users
✗ https://example.com/api/internal/debug

Should You Always Follow robots.txt?

robots.txt is advisory, not legally binding in most jurisdictions. But:

Follow it for sites you have no business relationship with
Respect Crawl-delay — it's telling you their server's capacity
Document your decisions — if you choose to ignore specific rules, have a reason

Rate Limiting: Don't Be That Scraper

The Golden Rule

Your scraper should be indistinguishable from a human browsing the site. A human doesn't load 100 pages per second.

Implementing Respectful Rate Limiting

import asyncio
import time
from collections import defaultdict

class RespectfulRateLimiter:
    """Rate limiter that respects site capacity."""

    def __init__(
        self,
        default_delay: float = 2.0,
        max_concurrent: int = 3,
    ):
        self.default_delay = default_delay
        self.max_concurrent = max_concurrent
        self._domain_semaphores: dict[str, asyncio.Semaphore] = {}
        self._last_request: dict[str, float] = defaultdict(float)
        self._lock = asyncio.Lock()

    def _get_semaphore(self, domain: str) -> asyncio.Semaphore:
        if domain not in self._domain_semaphores:
            self._domain_semaphores[domain] = asyncio.Semaphore(
                self.max_concurrent
            )
        return self._domain_semaphores[domain]

    async def acquire(self, domain: str, crawl_delay: float = None):
        """Wait for permission to make a request."""
        sem = self._get_semaphore(domain)
        await sem.acquire()

        delay = crawl_delay or self.default_delay

        async with self._lock:
            elapsed = time.monotonic() - self._last_request[domain]
            if elapsed < delay:
                await asyncio.sleep(delay - elapsed)
            self._last_request[domain] = time.monotonic()

    def release(self, domain: str):
        sem = self._get_semaphore(domain)
        sem.release()


# Usage with robots.txt
class ResponsibleScraper:
    def __init__(self):
        self.robots = RobotsChecker()
        self.limiter = RespectfulRateLimiter(
            default_delay=2.0,
            max_concurrent=3,
        )

    async def fetch(self, url: str) -> str | None:
        domain = urlparse(url).netloc

        # Step 1: Check robots.txt
        if not self.robots.can_fetch(url):
            print(f"Blocked by robots.txt: {url}")
            return None

        # Step 2: Respect crawl delay
        crawl_delay = self.robots.get_crawl_delay(domain)

        # Step 3: Rate limit
        await self.limiter.acquire(domain, crawl_delay)

        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(url)
                return resp.text
        finally:
            self.limiter.release(domain)

Adaptive Rate Limiting

Adjust your speed based on server response:

class AdaptiveRateLimiter:
    """Slow down when the server shows stress."""

    def __init__(self, base_delay: float = 1.0):
        self.base_delay = base_delay
        self.current_delay = base_delay
        self.max_delay = 30.0
        self._consecutive_errors = 0

    def record_response(
        self, status_code: int, response_time: float
    ):
        if status_code == 429:
            # Rate limited — back off significantly
            self.current_delay = min(
                self.current_delay * 3,
                self.max_delay
            )
            print(
                f"Rate limited! Delay → {self.current_delay:.1f}s"
            )

        elif status_code >= 500:
            # Server error — back off
            self._consecutive_errors += 1
            self.current_delay = min(
                self.base_delay * (2 ** self._consecutive_errors),
                self.max_delay
            )

        elif response_time > 5.0:
            # Slow response — server is struggling
            self.current_delay = min(
                self.current_delay * 1.5,
                self.max_delay
            )

        else:
            # Good response — gradually speed up
            self._consecutive_errors = 0
            self.current_delay = max(
                self.current_delay * 0.95,
                self.base_delay
            )

    async def wait(self):
        await asyncio.sleep(self.current_delay)

Identifying Your Scraper

Be transparent about who you are:

# Set a clear User-Agent that identifies your bot
HEADERS = {
    "User-Agent": (
        "MyCompanyScraper/1.0 "
        "(+https://mycompany.com/bot; "
        "contact@mycompany.com)"    ),
    "From": "contact@mycompany.com",}# This helps site owners:
# 1. Contact you if there's a problem
# 2. Whitelist you if they want to
# 3. Understand the traffic source

Handling CAPTCHAs Responsibly

When CAPTCHAs appear, they're a signal: the site wants to verify you're human. Options:

Option 1: Reduce Your Rate

async def handle_captcha_signal(scraper):
    """CAPTCHAs appearing = you're going too fast."""

    # First, slow down
    scraper.limiter.current_delay *= 2
    print(f"CAPTCHAs detected — slowing to "
          f"{scraper.limiter.current_delay:.1f}s/req")

    # If CAPTCHAs persist, solve them
    # But don't solve more than N per hour
    if scraper.captcha_count_this_hour < 50:
        token = await solver.solve(...)
        scraper.captcha_count_this_hour += 1
        return token
    else:
        print("Too many CAPTCHAs — stopping to avoid abuse")
        return None

Option 2: Solve When Necessary

For legitimate use cases (price monitoring, research, testing), solving CAPTCHAs is reasonable:

from datetime import datetime

class CaptchaBudget:
    """Track and limit CAPTCHA solving costs."""

    def __init__(
        self,
        daily_budget: float = 5.0,  # USD
        cost_per_solve: float = 0.001,
    ):
        self.daily_budget = daily_budget
        self.cost_per_solve = cost_per_solve
        self.today_spent = 0.0
        self.today_date = datetime.utcnow().date()

    def can_solve(self) -> bool:
        today = datetime.utcnow().date()
        if today != self.today_date:
            self.today_spent = 0.0
            self.today_date = today

        return self.today_spent + self.cost_per_solve <= self.daily_budget

    def record_solve(self):
        self.today_spent += self.cost_per_solve

    @property    def remaining(self) -> float:
        return self.daily_budget - self.today_spent

Caching: Don't Scrape What You Already Have

import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta

class ScrapeCache:
    """Cache scraped pages to avoid unnecessary requests."""

    def __init__(
        self,
        cache_dir: str = ".cache",
        ttl_hours: int = 24,
    ):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
        self.hits = 0
        self.misses = 0

    def _cache_key(self, url: str) -> str:
        return hashlib.md5(url.encode()).hexdigest()

    def get(self, url: str) -> str | None:
        key = self._cache_key(url)
        cache_file = self.cache_dir / f"{key}.json"

        if not cache_file.exists():
            self.misses += 1
            return None

        data = json.loads(cache_file.read_text())
        cached_at = datetime.fromisoformat(data["cached_at"])

        if datetime.utcnow() - cached_at > self.ttl:
            self.misses += 1
            return None

        self.hits += 1
        return data["html"]

    def set(self, url: str, html: str):
        key = self._cache_key(url)
        cache_file = self.cache_dir / f"{key}.json"
        cache_file.write_text(json.dumps({
            "url": url,
            "html": html,
            "cached_at": datetime.utcnow().isoformat(),
        }))

    @property    def hit_rate(self) -> str:
        total = self.hits + self.misses
        if total == 0:
            return "N/A"
        return f"{self.hits/total:.1%}"

The Complete Responsible Scraper

class ResponsibleScraper:
    def __init__(self, config: dict = None):
        config = config or {}

        self.robots = RobotsChecker()
        self.limiter = AdaptiveRateLimiter(
            base_delay=config.get("base_delay", 2.0)
        )
        self.cache = ScrapeCache(
            ttl_hours=config.get("cache_hours", 24)
        )
        self.captcha_budget = CaptchaBudget(
            daily_budget=config.get("daily_captcha_budget", 5.0)
        )
        self.captcha_solver = CaptchaSolver(
            api_base="https://www.passxapi.com"
        )
        self.stats = {
            "fetched": 0,
            "cached": 0,
            "robots_blocked": 0,
            "captchas_solved": 0,
            "rate_limited": 0,
        }

    async def scrape(self, url: str) -> dict | None:
        domain = urlparse(url).netloc

        # 1. Check cache
        cached = self.cache.get(url)
        if cached:
            self.stats["cached"] += 1
            return {"url": url, "html": cached, "cached": True}

        # 2. Check robots.txt
        if not self.robots.can_fetch(url):
            self.stats["robots_blocked"] += 1
            return None

        # 3. Rate limit
        crawl_delay = self.robots.get_crawl_delay(domain)
        await self.limiter.wait()

        # 4. Fetch
        async with httpx.AsyncClient(
            headers={
                "User-Agent": (
                    "DataCollector/1.0 "
                    "(+https://mysite.com/bot)"
                ),
            }
        ) as client:
            start = time.monotonic()
            resp = await client.get(url)
            elapsed = time.monotonic() - start

            # 5. Adapt rate based on response
            self.limiter.record_response(
                resp.status_code, elapsed
            )

            if resp.status_code == 429:
                self.stats["rate_limited"] += 1
                return None

            html = resp.text

            # 6. Handle CAPTCHA if present
            captcha = detect_captcha(html)
            if captcha:
                if self.captcha_budget.can_solve():
                    token = await self.captcha_solver.solve(
                        captcha_type=captcha["type"],
                        sitekey=captcha["sitekey"],
                        url=url,
                    )
                    self.captcha_budget.record_solve()
                    self.stats["captchas_solved"] += 1

                    # Resubmit with token
                    resp = await client.post(
                        url, data={captcha["field"]: token}
                    )
                    html = resp.text
                else:
                    print(
                        f"CAPTCHA budget exhausted "
                        f"(${self.captcha_budget.remaining:.2f} left)"
                    )
                    return None

            # 7. Cache the result
            self.cache.set(url, html)
            self.stats["fetched"] += 1

            return {"url": url, "html": html, "cached": False}

    def print_stats(self):
        print(f"Scraping stats: {self.stats}")
        print(f"Cache hit rate: {self.cache.hit_rate}")
        print(
            f"CAPTCHA budget remaining: "
            f"${self.captcha_budget.remaining:.2f}"
        )
        print(f"Current delay: {self.limiter.current_delay:.1f}s")

Quick Checklist

Before running your scraper in production:

[ ] robots.txt — Are you checking and respecting it?
[ ] Rate limiting — Are you waiting between requests?
[ ] User-Agent — Does it identify your bot and provide contact info?
[ ] Caching — Are you avoiding re-scraping unchanged pages?
[ ] Error handling — Do you back off on 429/5xx responses?
[ ] CAPTCHA budget — Have you set a daily spending limit?
[ ] Data storage — Are you only keeping data you actually need?
[ ] Terms of Service — Have you read the site's ToS?

Key Takeaways

robots.txt is your first check — respect it unless you have a documented reason not to
Adaptive rate limiting is better than fixed delays — respond to server signals
Cache aggressively — don't re-scrape what hasn't changed
Budget your CAPTCHA solves — set daily limits and stick to them
Identify yourself — a clear User-Agent helps everyone
Slow is reliable — a scraper that runs for a week at 1 req/s beats one that gets banned in an hour

For handling CAPTCHAs within your budget, check out passxapi-python — at $0.001/solve, even a $5/day budget gives you 5,000 solves.

What's your approach to responsible scraping? Share your practices in the comments.

The Complete Anti-Bot Detection Stack: Every Technique Sites Use to Catch Your Scraper

Alex Chen — Mon, 23 Mar 2026 01:30:13 +0000

You've handled CAPTCHAs, rotated proxies, and spoofed your User-Agent. Your scraper still gets blocked. Why?

Because modern anti-bot systems don't rely on a single detection method. They stack multiple layers — and you need to understand all of them to build a scraper that survives in production.

This guide maps out every detection technique sites use today, from basic to advanced.

The Detection Layers

Layer 7 ──── Application Logic
              ├── Rate limiting
              ├── Behavioral analysis
              └── Business logic traps

Layer 6 ──── CAPTCHA Challenges
              ├── reCAPTCHA v2/v3
              ├── hCaptcha / Enterprise
              ├── Cloudflare Turnstile
              └── FunCaptcha

Layer 5 ──── JavaScript Challenges
              ├── Browser fingerprinting
              ├── Canvas/WebGL hashing
              └── Proof-of-work puzzles

Layer 4 ──── HTTP Analysis
              ├── Header order/consistency
              ├── TLS fingerprint (JA3/JA4)
              └── HTTP/2 fingerprint

Layer 3 ──── Network Layer
              ├── IP reputation
              ├── ASN classification
              ├── Geo-location matching
              └── DNS analysis

Each layer catches different kinds of bots. Let's go through them all.

Layer 3: Network Analysis

IP Reputation

Anti-bot services maintain databases of known bad IPs:

import httpx

def check_ip_reputation(ip: str) -> dict:
    """Check if an IP is flagged in common databases."""

    checks = {}

    # AbuseIPDB
    resp = httpx.get(
        "https://api.abuseipdb.com/api/v2/check",
        params={"ipAddress": ip},
        headers={"Key": ABUSEIPDB_KEY}
    ).json()
    checks["abuse_score"] = resp["data"]["abuseConfidenceScore"]

    # Check if datacenter IP
    checks["is_datacenter"] = is_datacenter_ip(ip)

    return checks

def is_datacenter_ip(ip: str) -> bool:
    """Check if IP belongs to a known hosting provider."""
    import ipaddress

    DATACENTER_RANGES = [
        "13.0.0.0/8",     # AWS
        "34.0.0.0/8",     # GCP
        "40.0.0.0/8",     # Azure
        "104.16.0.0/12",  # Cloudflare
        "157.240.0.0/16", # Meta
    ]

    addr = ipaddress.ip_address(ip)
    return any(
        addr in ipaddress.ip_network(cidr)
        for cidr in DATACENTER_RANGES
    )

How to handle it:

Use residential proxies for sensitive targets
Rotate IPs from different subnets
Avoid datacenter IPs for sites with strict anti-bot

ASN Classification

Sites check your IP's Autonomous System Number (ASN) to identify hosting providers:

# What anti-bot services see:
# ✅ AS7922 (Comcast) → Residential
# ✅ AS7018 (AT&T) → Residential
# ❌ AS16509 (Amazon/AWS) → Datacenter
# ❌ AS14061 (DigitalOcean) → Datacenter
# ⚠️ AS13335 (Cloudflare) → CDN/VPN

Layer 4: HTTP Analysis

TLS Fingerprinting (JA3/JA4)

Every HTTP client has a unique TLS handshake signature:

# Different clients, different fingerprints:
# Chrome 121:  JA3 = 771,4865-4866-4867-49195...
# Python httpx: JA3 = 771,4866-4867-4865-49196...
# curl:         JA3 = 771,4865-4867-4866-49195...

# Solution: curl_cffi mimics browser TLS
from curl_cffi import requests

resp = requests.get(
    "https://target.com",
    impersonate="chrome120"
)
# Now your JA3 matches Chrome 120 exactly

Header Order

Browsers send headers in a specific order. Python libraries don't:

# Chrome sends:
# Host, Connection, sec-ch-ua, sec-ch-ua-mobile,
# sec-ch-ua-platform, Upgrade-Insecure-Requests,
# User-Agent, Accept, Sec-Fetch-Site...

# httpx sends:
# Host, User-Agent, Accept, Accept-Encoding,
# Connection...

# Fix: use OrderedDict or curl_cffi
headers = {
    "Host": "target.com",
    "Connection": "keep-alive",
    "sec-ch-ua": '"Chromium";v="121"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"Windows"',
    "Upgrade-Insecure-Requests": "1",
    "User-Agent": "Mozilla/5.0...",
    "Accept": "text/html,...",
}

HTTP/2 Fingerprinting

HTTP/2 settings (SETTINGS frame, WINDOW_UPDATE, PRIORITY) also create a unique fingerprint:

# Chrome's HTTP/2 settings:
# HEADER_TABLE_SIZE: 65536
# MAX_CONCURRENT_STREAMS: 1000
# INITIAL_WINDOW_SIZE: 6291456
# MAX_HEADER_LIST_SIZE: 262144

# Python's default: different values
# → Detectable

# curl_cffi handles this automatically
# when you use impersonate="chrome120"

Layer 5: JavaScript Challenges

Canvas Fingerprinting

Sites render hidden canvas elements and hash the result:

// What anti-bot JS does:
const canvas = document.createElement('canvas');
const ctx = canvas.getContext('2d');
ctx.textBaseline = "top";
ctx.font = "14px Arial";
ctx.fillText("Hello, world!", 2, 2);
const hash = canvas.toDataURL().hashCode();
// Different GPUs/drivers = different hash

WebGL Fingerprinting

const gl = canvas.getContext('webgl');
const debugInfo = gl.getExtension(
    'WEBGL_debug_renderer_info'
);

// Real browser:
gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL);
// → "ANGLE (NVIDIA GeForce GTX 1080 Ti...)"

// Headless Chrome:
// → "Google SwiftShader" ← DETECTED

Behavioral Analysis

Advanced systems track mouse movements, scroll patterns, and typing speed:

// What sites collect:
document.addEventListener('mousemove', (e) => {
    events.push({
        type: 'mouse',
        x: e.clientX, 
        y: e.clientY,
        t: Date.now()
    });
});

// Bot detection signals:
// - No mouse movement before form submit
// - Perfect straight-line mouse paths
// - Zero scroll events
// - Instant form fill (< 1 second)
// - No focus/blur events on inputs

How to simulate human behavior:

import random
import asyncio

async def human_like_interaction(page):
    """Simulate realistic user behavior."""

    # Random mouse movements
    for _ in range(random.randint(3, 7)):
        x = random.randint(100, 800)
        y = random.randint(100, 600)
        await page.mouse.move(x, y, steps=random.randint(5, 15))
        await asyncio.sleep(random.uniform(0.1, 0.5))

    # Scroll down naturally
    for _ in range(random.randint(2, 4)):
        await page.mouse.wheel(
            0, random.randint(100, 300)
        )
        await asyncio.sleep(random.uniform(0.5, 1.5))

    # Type with human-like delays
    async def human_type(selector, text):
        await page.click(selector)
        await asyncio.sleep(random.uniform(0.3, 0.8))
        for char in text:
            await page.keyboard.type(
                char, delay=random.randint(50, 150)
            )

Layer 6: CAPTCHA Challenges

The most visible layer. Sites deploy CAPTCHAs when other signals suggest bot activity:

# Unified CAPTCHA handler
async def handle_any_captcha(
    page, solver
) -> bool:
    """Detect and solve any CAPTCHA type."""

    detectors = {
        ".g-recaptcha": {
            "type": "recaptcha_v2",
            "key_attr": "data-sitekey"
        },
        ".h-captcha": {
            "type": "hcaptcha",
            "key_attr": "data-sitekey"
        },
        ".cf-turnstile": {
            "type": "turnstile",
            "key_attr": "data-sitekey"
        },
        "[data-pkey]": {
            "type": "funcaptcha",
            "key_attr": "data-pkey"
        },
    }

    for selector, config in detectors.items():
        el = await page.query_selector(selector)
        if el:
            sitekey = await el.get_attribute(
                config["key_attr"]
            )
            token = await solver.solve(
                captcha_type=config["type"],
                sitekey=sitekey,
                url=page.url,
            )
            await inject_token(page, config["type"], token)
            return True

    return False

Layer 7: Application Logic

Honeypot Fields

Hidden form fields that real users never fill:

<!-- Trap for bots -->
<input type="text" 
       name="website" 
       style="display:none" 
       tabindex="-1">

# Don't fill fields that are hidden!
async def fill_form_safely(page, data: dict):
    for field, value in data.items():
        el = await page.query_selector(
            f'input[name="{field}"]'
        )
        if el:
            visible = await el.is_visible()
            if visible:
                await el.fill(value)
            # Skip hidden fields — they're traps

Timing Analysis

Sites track how fast you interact:

# Too fast → bot
# A real user takes 5-30 seconds to fill a form
# A bot fills it in < 1 second

async def realistic_form_fill(page, data):
    # Wait before starting (reading the page)
    await asyncio.sleep(random.uniform(2, 5))

    for field, value in data.items():
        await page.click(f'input[name="{field}"]')
        await asyncio.sleep(random.uniform(0.5, 1.5))
        await page.type(
            f'input[name="{field}"]', value,
            delay=random.randint(30, 100)
        )
        await asyncio.sleep(random.uniform(0.3, 0.8))

    # Pause before submitting
    await asyncio.sleep(random.uniform(1, 3))

Building a Complete Anti-Detection Stack

class StealthScraper:
    """Scraper that handles all detection layers."""

    def __init__(self):
        self.proxy_pool = ResidentialProxyPool()  # Layer 3
        self.tls_client = TLSClient("chrome120")  # Layer 4
        self.browser_pool = BrowserPool(stealth=True)  # Layer 5
        self.captcha_solver = CaptchaSolver(  # Layer 6
            api_base="https://www.passxapi.com"
        )

    async def scrape(self, url: str) -> dict:
        # Layer 3: Get clean proxy
        proxy = await self.proxy_pool.get(
            region=get_target_region(url)
        )

        async with self.browser_pool.get_page(
            proxy=proxy
        ) as page:
            # Layer 5: Apply stealth patches
            await apply_stealth(page)

            # Layer 7: Simulate human behavior
            await human_like_interaction(page)

            # Navigate
            await page.goto(url)

            # Layer 6: Handle CAPTCHAs
            await handle_any_captcha(
                page, self.captcha_solver
            )

            # Layer 7: Avoid honeypots
            await fill_form_safely(page, form_data)

            return await extract_data(page)

Detection Layer Priorities

Not all layers matter equally. Focus on what catches you:

Layer	Detection Rate	Effort to Bypass	Priority
IP reputation	30%	Low (proxies)	High
TLS fingerprint	25%	Low (curl_cffi)	High
CAPTCHA	20%	Medium (API solver)	High
JS fingerprint	15%	Medium (stealth)	Medium
Behavioral	5%	High (simulation)	Low
Honeypots	3%	Low (skip hidden)	Low
Header order	2%	Low (manual)	Low

Start from the top. Most scrapers get blocked at layers 3-4, not 5-7.

Key Takeaways

Detection is layered — fixing one layer while ignoring others won't work
Start with network + TLS — these catch 55% of bots before JS even runs
CAPTCHAs are the visible layer — but they're triggered by invisible signals
Behavioral analysis is growing — mouse movement and timing matter more each year
Test your stealth — use bot detection sites to audit your setup
Always have a CAPTCHA solver — even perfect stealth can't avoid all challenges

For handling the CAPTCHA layer when it triggers, check out passxapi-python — it provides a unified API for reCAPTCHA, hCaptcha, Turnstile, and FunCaptcha, so you can focus on the other layers.

Which detection layer causes you the most trouble? Share your experience in the comments.

reCAPTCHA v2 vs v3: What's Different, What Breaks Your Scraper, and How to Handle Both

Alex Chen — Mon, 23 Mar 2026 01:23:37 +0000

reCAPTCHA v2 shows a checkbox. reCAPTCHA v3 is invisible. But the differences go much deeper than that — and understanding them determines whether your scraper works or silently fails.

Let's break down how each version works, how sites implement them, and how to handle both in your automation.

The Fundamental Difference

reCAPTCHA v2: Binary — you either pass or fail. The user sees a challenge (checkbox, image selection) and the server gets a pass/fail token.

reCAPTCHA v3: Score-based — every request gets a risk score from 0.0 (bot) to 1.0 (human). The site owner decides the threshold. No user interaction.

v2 Flow:
User → Checkbox → (maybe image puzzle) → Token → Server verifies → Pass/Fail

v3 Flow:
Page loads → JS collects signals → Score assigned (0.0-1.0) → Token → Server checks score

How v2 Works

The Checkbox ("I'm not a robot")

When a user clicks the checkbox, Google's JavaScript:

Analyzes mouse movement patterns to the checkbox
Checks browser fingerprint
Reviews cookies and browsing history
Makes a risk decision

If low risk → green checkmark, token issued.
If uncertain → image challenge appears.

<!-- v2 HTML embed -->
<div class="g-recaptcha" 
     data-sitekey="6Le-wvkSAAAAAPBM..."
     data-callback="onCaptchaSuccess">
</div>
<script src="https://www.google.com/recaptcha/api.js" 
        async defer></script>

The Hidden Input

After solving, v2 populates a hidden textarea:

<textarea name="g-recaptcha-response" 
          style="display:none">
03AGdBq24P...long-token...
</textarea>

Server-Side Verification

# What the site's backend does
import httpx

def verify_recaptcha_v2(token: str) -> bool:
    resp = httpx.post(
        "https://www.google.com/recaptcha/api/siteverify",
        data={
            "secret": RECAPTCHA_SECRET_KEY,
            "response": token,
        }
    ).json()

    return resp["success"]  # True or False
    # That's it — binary result

How v3 Works

Invisible Scoring

v3 runs entirely in the background. The site loads a script, calls execute()\ on user actions, and gets a score:

<!-- v3 HTML embed -->
<script src="https://www.google.com/recaptcha/api.js?render=SITE_KEY">
</script>
<script>
  grecaptcha.ready(function() {
    // Called on every important action
    grecaptcha.execute('SITE_KEY', {action: 'login'})
      .then(function(token) {
        // Send token to your server
        document.getElementById('captcha-token')
          .value = token;
      });
  });
</script>

The Score Response

# v3 server-side verification
def verify_recaptcha_v3(
    token: str, 
    expected_action: str = "login"
) -> dict:
    resp = httpx.post(
        "https://www.google.com/recaptcha/api/siteverify",
        data={
            "secret": RECAPTCHA_SECRET_KEY,
            "response": token,
        }
    ).json()

    return {
        "success": resp["success"],
        "score": resp.get("score", 0),
        # 1.0 = very likely human
        # 0.0 = very likely bot
        "action": resp.get("action", ""),
        "hostname": resp.get("hostname", ""),
    }

# Site owner decides the threshold
result = verify_recaptcha_v3(token)
if result["score"] < 0.5:
    # Too risky — show v2 fallback or block
    show_recaptcha_v2()
elif result["action"] != "login":
    # Action mismatch — possible replay
    block_request()

Side-by-Side Comparison

Feature	v2	v3
User interaction	Checkbox + puzzle	None (invisible)
Result type	Pass/Fail	Score (0.0-1.0)
Threshold	Fixed	Site-configurable
Action parameter	No	Yes
Multiple calls/page	Usually once	Multiple (per action)
Sitekey location	`data-sitekey\` attribute	Script URL or JS call
Token field	`g-recaptcha-response\`	Custom (varies)
Fallback	Image challenge	Often falls back to v2

Detecting Which Version You're Facing

import re
from bs4 import BeautifulSoup

def detect_recaptcha_version(html: str) -> dict | None:
    soup = BeautifulSoup(html, "html.parser")

    # Check for v2 — explicit checkbox widget
    v2_div = soup.find("div", class_="g-recaptcha")
    if v2_div:
        return {
            "version": "v2",
            "sitekey": v2_div.get("data-sitekey"),
            "type": (
                "invisible" 
                if v2_div.get("data-size") == "invisible"
                else "checkbox"
            ),
        }

    # Check for v3 — render parameter in script
    for script in soup.find_all("script", src=True):
        src = script["src"]
        if "recaptcha" in src and "render=" in src:
            match = re.search(r'render=([^&"]+)', src)
            if match and match.group(1) != "explicit":
                return {
                    "version": "v3",
                    "sitekey": match.group(1),
                    "type": "score",
                }

    # Check for v3 — grecaptcha.execute in inline JS
    for script in soup.find_all("script"):
        if script.string and "grecaptcha.execute" in script.string:
            match = re.search(
                r"grecaptcha\.execute\(['\"]([^'\"]+)",
                script.string
            )
            if match:
                return {
                    "version": "v3",
                    "sitekey": match.group(1),
                    "type": "score",
                }

    return None

Solving reCAPTCHA v2

v2 is straightforward — you need a valid token:

import httpx
import time

def solve_recaptcha_v2(
    sitekey: str, 
    page_url: str,
    invisible: bool = False
) -> str:
    client = httpx.Client(
        base_url="https://www.passxapi.com"
    )

    payload = {
        "type": "recaptcha_v2",
        "sitekey": sitekey,
        "pageurl": page_url,
    }
    if invisible:
        payload["invisible"] = True

    # Submit task
    task = client.post(
        "/api/v1/task", json=payload
    ).json()
    task_id = task["task_id"]

    # Poll for result
    for _ in range(60):
        result = client.get(
            f"/api/v1/task/{task_id}"
        ).json()
        if result["status"] == "completed":
            return result["token"]
        time.sleep(2)

    raise TimeoutError("v2 solve timed out")

Injecting the v2 Token

# With Playwright
async def inject_v2_token(page, token: str):
    await page.evaluate(f"""() => {{
        document.querySelector(
            '#g-recaptcha-response'
        ).value = '{token}';

        // Also set in all iframes
        document.querySelectorAll(
            'textarea[name="g-recaptcha-response"]'
        ).forEach(el => el.value = '{token}');

        // Trigger the callback if it exists
        const callback = document.querySelector(
            '.g-recaptcha'
        )?.getAttribute('data-callback');
        if (callback && window[callback]) {{
            window[callback]('{token}');
        }}
    }}""")

# Without browser — just POST
def submit_with_v2_token(
    url: str, token: str, form_data: dict
):
    form_data["g-recaptcha-response"] = token
    return httpx.post(url, data=form_data)

Solving reCAPTCHA v3

v3 requires an extra parameter — the action string:

def solve_recaptcha_v3(
    sitekey: str,
    page_url: str,
    action: str = "submit",
    min_score: float = 0.7,
) -> str:
    client = httpx.Client(
        base_url="https://www.passxapi.com"
    )

    task = client.post("/api/v1/task", json={
        "type": "recaptcha_v3",
        "sitekey": sitekey,
        "pageurl": page_url,
        "action": action,
        "min_score": min_score,
    }).json()
    task_id = task["task_id"]

    for _ in range(60):
        result = client.get(
            f"/api/v1/task/{task_id}"
        ).json()
        if result["status"] == "completed":
            return result["token"]
        time.sleep(2)

    raise TimeoutError("v3 solve timed out")

Finding the Action String

The action string must match what the site expects. Find it in the page source:

import re

def extract_v3_action(html: str) -> str | None:
    """Find the action parameter from page JS."""
    patterns = [
        r"grecaptcha\.execute\([^,]+,\s*\{action:\s*['\"]([^'\"]+)",
        r"action\s*:\s*['\"]([^'\"]+)['\"]",
    ]

    for pattern in patterns:
        match = re.search(pattern, html)
        if match:
            return match.group(1)

    return None

# Common action strings
COMMON_ACTIONS = [
    "submit", "login", "register", "signup",
    "contact", "checkout", "search", "homepage",
    "validate", "verify",
]

Injecting the v3 Token

v3 tokens are often sent differently — in headers, hidden fields, or via AJAX:

# Method 1: Hidden input
async def inject_v3_hidden_input(page, token):
    await page.evaluate(f"""() => {{
        // Find the hidden input for the token
        let input = document.querySelector(
            'input[name="recaptcha_token"], '
            + 'input[name="g-recaptcha-response"], '
            + 'input[name="captcha_token"]'
        );
        if (!input) {{
            // Create it if it doesn't exist
            input = document.createElement('input');
            input.type = 'hidden';
            input.name = 'g-recaptcha-response';
            document.querySelector('form')
                .appendChild(input);
        }}
        input.value = '{token}';
    }}""")

# Method 2: Direct POST with token
def submit_v3_form(
    url: str, token: str, 
    form_data: dict, 
    token_field: str = "g-recaptcha-response"
):
    form_data[token_field] = token
    return httpx.post(url, data=form_data)

# Method 3: API call with token in headers
def api_call_with_v3(
    url: str, token: str, payload: dict
):
    return httpx.post(
        url,
        json=payload,
        headers={"X-Recaptcha-Token": token}
    )

The v3 → v2 Fallback Pattern

Many sites use v3 first, then show v2 if the score is low:

async def handle_recaptcha_adaptive(
    page, sitekey: str, url: str
):
    """Handle sites that use v3 with v2 fallback."""

    # Try v3 first
    v3_token = solve_recaptcha_v3(
        sitekey=sitekey,
        page_url=url,
        action=extract_v3_action(
            await page.content()
        ) or "submit",
        min_score=0.9,
    )

    await inject_v3_hidden_input(page, v3_token)
    await page.click('button[type="submit"]')
    await page.wait_for_timeout(2000)

    # Check if v2 appeared as fallback
    v2_widget = await page.query_selector(
        ".g-recaptcha:visible, "
        "#recaptcha-anchor:visible"
    )

    if v2_widget:
        print("v3 score too low — falling back to v2")
        v2_sitekey = await v2_widget.get_attribute(
            "data-sitekey"
        ) or sitekey

        v2_token = solve_recaptcha_v2(
            sitekey=v2_sitekey,
            page_url=url,
        )

        await inject_v2_token(page, v2_token)
        await page.click('button[type="submit"]')

Common Pitfalls

1. Wrong Action String for v3

# ❌ Generic action
token = solve_v3(sitekey, url, action="submit")

# ✅ Match the site's actual action
action = extract_v3_action(page_html) or "login"
token = solve_v3(sitekey, url, action=action)

If the action doesn't match, the server rejects the token even if the score is high.

2. Confusing v2 Invisible with v3

# v2 invisible — still has a widget, just auto-triggered
# data-size="invisible" on the div
<div class="g-recaptcha" 
     data-sitekey="..." 
     data-size="invisible">

# v3 — loaded via render parameter, no div at all
<script src="...recaptcha/api.js?render=SITEKEY">

v2 invisible needs to be solved as v2 (with invisible=True\), not v3.

3. Token Field Names Vary

# Common field names for v3 tokens
V3_FIELD_NAMES = [
    "g-recaptcha-response",
    "recaptcha_token",
    "captcha_token",
    "recaptchaToken",
    "token",
    "grecaptcha",
]

def find_token_field(html: str) -> str:
    for field in V3_FIELD_NAMES:
        if f'name="{field}"' in html:
            return field
    return "g-recaptcha-response"  # Default

Unified Solver

Handle both versions with a single function:

async def solve_recaptcha(
    html: str, 
    url: str
) -> dict | None:
    info = detect_recaptcha_version(html)
    if not info:
        return None

    if info["version"] == "v2":
        token = solve_recaptcha_v2(
            sitekey=info["sitekey"],
            page_url=url,
            invisible=info["type"] == "invisible",
        )
        return {
            "token": token,
            "field": "g-recaptcha-response",
            "version": "v2",
        }

    elif info["version"] == "v3":
        action = extract_v3_action(html) or "submit"
        token = solve_recaptcha_v3(
            sitekey=info["sitekey"],
            page_url=url,
            action=action,
        )
        field = find_token_field(html)
        return {
            "token": token,
            "field": field,
            "version": "v3",
            "action": action,
        }

Key Takeaways

v2 is binary, v3 is scored — different handling required
v3 needs the action string — extract it from page JS, don't guess
v2 invisible ≠ v3 — same appearance, different protocol
Many sites use v3→v2 fallback — handle both in sequence
Token field names vary — especially for v3, check the form
Test with Google's test keys — v2: 6LeIxAcTAAAA...\, always passes

For a Python client that handles both v2 and v3 with a unified API, check out passxapi-python — it abstracts away the version differences so your code stays clean.

Which version gives you more trouble — v2 or v3? Share your experience in the comments.

Managing Browser Pools for Large-Scale Scraping: Memory, Crashes, and CAPTCHA Handling

Alex Chen — Mon, 23 Mar 2026 01:15:27 +0000

You spin up 20 Playwright browsers for parallel scraping. After an hour, your server runs out of memory. Or one browser crashes and takes your whole pipeline with it.

Browser-based scraping at scale needs more than just browser.launch()\. You need a browser pool — with lifecycle management, memory limits, crash recovery, and integrated CAPTCHA handling.

Why You Need a Browser Pool

Single browser problems:

Memory leaks — Chrome eats ~100-300MB per instance, pages leak more
Zombie processes — crashed browsers leave orphan processes
No concurrency control — 50 pages in one browser = tab soup
No isolation — one bad page crashes everything

A pool solves all of these.

Step 1: The Basic Pool

# browser_pool.py
import asyncio
from playwright.async_api import async_playwright
from dataclasses import dataclass, field
from datetime import datetime
import os

@dataclassclass BrowserInstance:
    browser: object
    context: object
    created_at: datetime = field(
        default_factory=datetime.utcnow
    )
    pages_processed: int = 0
    is_busy: bool = False
    pid: int = 0

class BrowserPool:
    def __init__(
        self,
        size: int = 5,
        max_pages_per_browser: int = 50,
        max_age_seconds: int = 300,
    ):
        self.size = size
        self.max_pages = max_pages_per_browser
        self.max_age = max_age_seconds
        self._pool: list[BrowserInstance] = []
        self._lock = asyncio.Lock()
        self._playwright = None
        self._semaphore = asyncio.Semaphore(size)

    async def start(self):
        self._playwright = await async_playwright().start()
        for _ in range(self.size):
            instance = await self._create_instance()
            self._pool.append(instance)

    async def _create_instance(self) -> BrowserInstance:
        browser = await self._playwright.chromium.launch(
            headless=True,
            args=[
                "--disable-dev-shm-usage",
                "--disable-gpu",
                "--no-sandbox",
                "--disable-setuid-sandbox",
                # Limit memory per process
                "--max-old-space-size=256",
                "--disable-blink-features="
                "AutomationControlled",
            ]
        )

        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/121.0.0.0 Safari/537.36"
            ),
        )

        # Get browser PID for monitoring
        pid = browser.process.pid if browser.process else 0

        return BrowserInstance(
            browser=browser,
            context=context,
            pid=pid,
        )

    async def acquire(self) -> BrowserInstance:
        """Get a browser from the pool."""
        await self._semaphore.acquire()

        async with self._lock:
            for instance in self._pool:
                if not instance.is_busy:
                    # Check if browser needs recycling
                    if self._needs_recycle(instance):
                        await self._recycle(instance)

                    instance.is_busy = True
                    return instance

        # Shouldn't reach here with semaphore
        raise RuntimeError("No available browsers")

    async def release(self, instance: BrowserInstance):
        """Return a browser to the pool."""
        async with self._lock:
            instance.is_busy = False
            instance.pages_processed += 1
        self._semaphore.release()

    def _needs_recycle(self, instance: BrowserInstance) -> bool:
        age = (
            datetime.utcnow() - instance.created_at
        ).total_seconds()
        return (
            instance.pages_processed >= self.max_pages
            or age > self.max_age
        )

    async def _recycle(self, instance: BrowserInstance):
        """Close old browser and create a fresh one."""
        idx = self._pool.index(instance)

        try:
            await instance.context.close()
            await instance.browser.close()
        except Exception:
            pass  # Browser may already be dead

        new_instance = await self._create_instance()
        self._pool[idx] = new_instance

    async def shutdown(self):
        """Close all browsers."""
        for instance in self._pool:
            try:
                await instance.context.close()
                await instance.browser.close()
            except Exception:
                pass

        if self._playwright:
            await self._playwright.stop()

Step 2: Context Manager for Safe Usage

from contextlib import asynccontextmanager

class BrowserPool:
    # ... (previous code) ...

    @asynccontextmanager    async def get_page(self):
        """Get a page from the pool, 
        auto-release on exit."""
        instance = await self.acquire()
        page = None

        try:
            page = await instance.context.new_page()
            yield page
        except Exception as e:
            # Browser might have crashed
            if self._is_crash(e):
                await self._handle_crash(instance)
            raise
        finally:
            if page:
                try:
                    await page.close()
                except Exception:
                    pass
            await self.release(instance)

    def _is_crash(self, error: Exception) -> bool:
        crash_indicators = [
            "Target closed",
            "Browser closed",
            "Protocol error",
            "Connection refused",
        ]
        return any(
            indicator in str(error) 
            for indicator in crash_indicators
        )

    async def _handle_crash(
        self, instance: BrowserInstance
    ):
        """Replace crashed browser with a new one."""
        async with self._lock:
            try:
                idx = self._pool.index(instance)
                # Kill zombie process if needed
                if instance.pid:
                    try:
                        os.kill(instance.pid, 9)
                    except ProcessLookupError:
                        pass

                new_instance = await self._create_instance()
                self._pool[idx] = new_instance
                print(
                    f"Replaced crashed browser "
                    f"(PID {instance.pid})"
                )
            except ValueError:
                pass  # Already replaced

Step 3: Memory Monitoring

Chrome is a memory hog. Monitor and kill browsers that get too fat:

import psutil

class MemoryMonitor:
    def __init__(
        self, 
        pool: BrowserPool,
        max_memory_mb: int = 500,
        check_interval: int = 30,
    ):
        self.pool = pool
        self.max_memory = max_memory_mb * 1024 * 1024
        self.interval = check_interval
        self._task = None

    async def start(self):
        self._task = asyncio.create_task(
            self._monitor_loop()
        )

    async def _monitor_loop(self):
        while True:
            await asyncio.sleep(self.interval)
            await self._check_memory()

    async def _check_memory(self):
        for instance in self.pool._pool:
            if not instance.pid:
                continue

            try:
                proc = psutil.Process(instance.pid)
                memory = proc.memory_info().rss

                # Include child processes
                for child in proc.children(recursive=True):
                    memory += child.memory_info().rss

                memory_mb = memory / (1024 * 1024)

                if memory > self.max_memory:
                    print(
                        f"Browser PID {instance.pid} "
                        f"using {memory_mb:.0f}MB "
                        f"(limit: "
                        f"{self.max_memory/1024/1024:.0f}MB)"
                        f" — recycling"
                    )
                    if not instance.is_busy:
                        await self.pool._recycle(instance)

            except (
                psutil.NoSuchProcess, 
                psutil.AccessDenied
            ):
                pass

    async def get_stats(self) -> dict:
        total_memory = 0
        stats = []

        for i, instance in enumerate(self.pool._pool):
            try:
                proc = psutil.Process(instance.pid)
                mem = proc.memory_info().rss
                for child in proc.children(recursive=True):
                    mem += child.memory_info().rss

                total_memory += mem
                stats.append({
                    "id": i,
                    "pid": instance.pid,
                    "memory_mb": mem / (1024 * 1024),
                    "pages": instance.pages_processed,
                    "busy": instance.is_busy,
                })
            except psutil.NoSuchProcess:
                stats.append({
                    "id": i, "status": "dead"
                })

        return {
            "browsers": stats,
            "total_memory_mb": total_memory / (1024*1024),
            "pool_size": self.pool.size,
        }

Step 4: CAPTCHA-Aware Scraper Using the Pool

# scraper.py
from browser_pool import BrowserPool, MemoryMonitor

class PooledScraper:
    def __init__(
        self,
        pool_size: int = 5,
        captcha_solver = None,
    ):
        self.pool = BrowserPool(
            size=pool_size,
            max_pages_per_browser=50,
            max_age_seconds=300,
        )
        self.monitor = MemoryMonitor(
            self.pool, max_memory_mb=500
        )
        self.solver = captcha_solver
        self.stats = {
            "scraped": 0, "captchas": 0, 
            "errors": 0, "crashes": 0,
        }

    async def start(self):
        await self.pool.start()
        await self.monitor.start()

    async def scrape(self, url: str) -> dict:
        async with self.pool.get_page() as page:
            try:
                await page.goto(
                    url, 
                    wait_until="domcontentloaded",
                    timeout=30000
                )

                # Check for CAPTCHA
                captcha = await self._detect_captcha(page)

                if captcha and self.solver:
                    await self._solve_captcha(
                        page, captcha
                    )
                    self.stats["captchas"] += 1

                # Extract data
                content = await page.content()
                self.stats["scraped"] += 1

                return {
                    "url": url,
                    "html": content,
                    "captcha_solved": bool(captcha),
                }

            except Exception as e:
                self.stats["errors"] += 1
                return {"url": url, "error": str(e)}

    async def _detect_captcha(self, page) -> dict | None:
        selectors = {
            "recaptcha_v2": ".g-recaptcha",
            "hcaptcha": ".h-captcha",
            "turnstile": ".cf-turnstile",
        }

        for ctype, selector in selectors.items():
            el = await page.query_selector(selector)
            if el:
                sitekey = await el.get_attribute(
                    "data-sitekey"
                )
                return {
                    "type": ctype, 
                    "sitekey": sitekey
                }

        return None

    async def _solve_captcha(self, page, captcha):
        token = await self.solver.solve(
            captcha_type=captcha["type"],
            sitekey=captcha["sitekey"],
            url=page.url,
        )

        # Inject token
        field_map = {
            "recaptcha_v2": "g-recaptcha-response",
            "hcaptcha": "h-captcha-response",
            "turnstile": "cf-turnstile-response",
        }
        field = field_map.get(
            captcha["type"], "captcha-response"
        )

        await page.evaluate(f"""() => {{
            const el = document.querySelector(
                'textarea[name="{field}"]'
            );
            if (el) el.value = '{token}';
        }}""")

    async def scrape_many(
        self, urls: list[str]
    ) -> list[dict]:
        tasks = [self.scrape(url) for url in urls]
        return await asyncio.gather(*tasks)

    async def shutdown(self):
        mem_stats = await self.monitor.get_stats()
        print(f"Memory: {mem_stats}")
        print(f"Scraper stats: {self.stats}")
        await self.pool.shutdown()

Step 5: Using It

import asyncio

async def main():
    # Setup
    solver = CaptchaSolver(
        api_base="https://www.passxapi.com"
    )
    scraper = PooledScraper(
        pool_size=5,
        captcha_solver=solver,
    )
    await scraper.start()

    # Load URLs
    urls = [
        f"https://shop.example.com/product/{i}"
        for i in range(1, 1001)
    ]

    # Scrape in batches
    batch_size = 20
    all_results = []

    for i in range(0, len(urls), batch_size):
        batch = urls[i:i + batch_size]
        results = await scraper.scrape_many(batch)
        all_results.extend(results)

        print(
            f"Batch {i//batch_size + 1}: "
            f"{scraper.stats}"
        )

    # Cleanup
    await scraper.shutdown()

    # Results
    success = [
        r for r in all_results if "html" in r
    ]
    print(
        f"Done: {len(success)}/{len(urls)} succeeded"
    )

asyncio.run(main())

Resource Limits for Docker

When running in containers, set hard limits:

# docker-compose.yml
services:
  scraper:
    build: .
    deploy:
      resources:
        limits:
          memory: 2G  # Hard cap
          cpus: "2.0"
        reservations:
          memory: 512M
    environment:
      - POOL_SIZE=5
      - MAX_MEMORY_PER_BROWSER=400
    # Shared memory for Chrome
    shm_size: "1gb"

Without shm_size\, Chrome will crash with "Out of shared memory" errors.

Health Endpoint

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")async def health():
    mem = await scraper.monitor.get_stats()
    return {
        "status": "healthy",
        "pool": {
            "size": scraper.pool.size,
            "total_memory_mb": mem["total_memory_mb"],
        },
        "scraper": scraper.stats,
    }

Key Takeaways

Pool browsers, don't spawn per-request — reuse reduces overhead from 2s to 50ms
Recycle after N pages or N minutes — prevents memory leaks
Monitor memory per-process — kill fat browsers before OOM
Handle crashes gracefully — replace dead browsers, don't restart everything
Use context managers — auto-release on success or failure
Set Docker shm_size — Chrome needs shared memory
CAPTCHA solving is just another step — detect, solve via API, inject, continue

For the CAPTCHA-solving integration, check out passxapi-python — it provides async clients that work seamlessly inside browser pool workflows.

How many concurrent browsers do you run? What's your memory limit? Share your setup in the comments.

Building an ETL Pipeline for Web Scraping: From CAPTCHA-Protected Pages to Clean Data

Alex Chen — Mon, 23 Mar 2026 01:08:18 +0000

Most scraping tutorials stop at "get the HTML." But in production, getting the HTML is just step one. You need to detect CAPTCHAs, solve them, extract structured data, validate it, handle duplicates, and load it into a database — reliably, at scale.

This is an ETL (Extract, Transform, Load) pipeline. Let's build one.

The Architecture

┌─────────────┐    ┌──────────────┐    ┌────────────┐
│  URL Queue   │───▶│  Fetcher     │───▶│  CAPTCHA   │
│  (Redis)     │    │  (aiohttp)   │    │  Detector  │
└─────────────┘    └──────────────┘    └────────────┘
                                             │
                              ┌──────────────┘
                              ▼
                    ┌──────────────┐    ┌────────────┐
                    │  CAPTCHA     │───▶│  Parser    │
                    │  Solver      │    │  (BS4)     │
                    └──────────────┘    └────────────┘
                                             │
                              ┌──────────────┘
                              ▼
                    ┌──────────────┐    ┌────────────┐
                    │  Validator   │───▶│  Deduper   │
                    │  (Pydantic)  │    │  (bloom)   │
                    └──────────────┘    └────────────┘
                                             │
                              ┌──────────────┘
                              ▼
                    ┌──────────────┐
                    │  Loader      │
                    │  (Postgres)  │
                    └──────────────┘

Step 1: Define Your Data Model

Start with what you want to end up with:

# models.py
from pydantic import BaseModel, HttpUrl, field_validator
from datetime import datetime
from decimal import Decimal

class Product(BaseModel):
    url: HttpUrl
    title: str
    price: Decimal
    currency: str = "USD"
    in_stock: bool
    description: str | None = None
    image_url: HttpUrl | None = None
    scraped_at: datetime
    source_domain: str

    @field_validator("title")    @classmethod    def title_not_empty(cls, v):
        v = v.strip()
        if len(v) < 3:
            raise ValueError("Title too short")
        return v

    @field_validator("price")    @classmethod    def price_positive(cls, v):
        if v <= 0:
            raise ValueError("Price must be positive")
        return v

class ScrapedPage(BaseModel):
    url: str
    html: str
    status_code: int
    captcha_solved: bool = False
    captcha_type: str | None = None
    fetch_time: float
    timestamp: datetime

Step 2: The Fetcher with CAPTCHA Handling

# fetcher.py
import aiohttp
import asyncio
import time
from dataclasses import dataclass, field

@dataclassclass FetchResult:
    url: str
    html: str = ""
    status: int = 0
    captcha_type: str | None = None
    captcha_solved: bool = False
    error: str | None = None
    fetch_time: float = 0

class Fetcher:
    def __init__(
        self,
        max_concurrent: int = 20,
        captcha_solver = None,
    ):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.solver = captcha_solver
        self.stats = {
            "fetched": 0, "captchas": 0, "errors": 0
        }

    async def fetch(
        self, 
        session: aiohttp.ClientSession,
        url: str
    ) -> FetchResult:
        async with self.semaphore:
            start = time.monotonic()
            try:
                async with session.get(
                    url, timeout=aiohttp.ClientTimeout(total=30)
                ) as resp:
                    html = await resp.text()
                    fetch_time = time.monotonic() - start

                    # Detect CAPTCHA
                    captcha = detect_captcha(html, url)

                    if captcha and self.solver:
                        token = await self.solver.solve(
                            captcha_type=captcha["type"],
                            sitekey=captcha["sitekey"],
                            url=url
                        )

                        if token:
                            # Resubmit with token
                            async with session.post(
                                url,
                                data={
                                    captcha["field"]: token
                                }
                            ) as resp2:
                                html = await resp2.text()
                                self.stats["captchas"] += 1

                    self.stats["fetched"] += 1
                    return FetchResult(
                        url=url,
                        html=html,
                        status=resp.status,
                        captcha_type=(
                            captcha["type"] if captcha 
                            else None
                        ),
                        captcha_solved=bool(captcha),
                        fetch_time=fetch_time,
                    )

            except Exception as e:
                self.stats["errors"] += 1
                return FetchResult(
                    url=url,
                    error=str(e),
                    fetch_time=time.monotonic() - start,
                )


def detect_captcha(html: str, url: str) -> dict | None:
    """Detect CAPTCHA type and extract sitekey."""
    import re

    patterns = {
        "recaptcha_v2": {
            "regex": r'class="g-recaptcha"[^>]*data-sitekey="([^"]+)"',
            "field": "g-recaptcha-response",
        },
        "hcaptcha": {
            "regex": r'class="h-captcha"[^>]*data-sitekey="([^"]+)"',
            "field": "h-captcha-response",
        },
        "turnstile": {
            "regex": r'class="cf-turnstile"[^>]*data-sitekey="([^"]+)"',
            "field": "cf-turnstile-response",
        },
    }

    for ctype, config in patterns.items():
        match = re.search(config["regex"], html)
        if match:
            return {
                "type": ctype,
                "sitekey": match.group(1),
                "field": config["field"],
            }

    return None

Step 3: The Parser

Extract structured data from raw HTML:

# parser.py
from bs4 import BeautifulSoup
from decimal import Decimal, InvalidOperation
from datetime import datetime
import re

class ProductParser:
    """Parse product data from HTML.
    Override for different site structures."""

    def parse(self, html: str, url: str) -> dict | None:
        soup = BeautifulSoup(html, "lxml")

        try:
            return {
                "url": url,
                "title": self._extract_title(soup),
                "price": self._extract_price(soup),
                "currency": self._extract_currency(soup),
                "in_stock": self._extract_stock(soup),
                "description": self._extract_description(soup),
                "image_url": self._extract_image(soup),
                "scraped_at": datetime.utcnow(),
                "source_domain": self._get_domain(url),
            }
        except Exception as e:
            print(f"Parse error for {url}: {e}")
            return None

    def _extract_title(self, soup) -> str:
        selectors = [
            "h1.product-title",
            "h1[itemprop='name']",
            ".product-name h1",
            "h1",
        ]
        for sel in selectors:
            el = soup.select_one(sel)
            if el and el.text.strip():
                return el.text.strip()
        raise ValueError("No title found")

    def _extract_price(self, soup) -> Decimal:
        selectors = [
            "[itemprop='price']",
            ".price-current",
            ".product-price",
            "[data-price]",
        ]
        for sel in selectors:
            el = soup.select_one(sel)
            if el:
                # Try content attribute first
                price_str = (
                    el.get("content") 
                    or el.get("data-price") 
                    or el.text
                )
                price_str = re.sub(
                    r'[^\d.]', '', price_str
                )
                try:
                    return Decimal(price_str)
                except InvalidOperation:
                    continue
        raise ValueError("No price found")

    def _extract_stock(self, soup) -> bool:
        # Check for "out of stock" indicators
        oos_indicators = [
            ".out-of-stock",
            "[data-availability='OutOfStock']",
        ]
        for sel in oos_indicators:
            if soup.select_one(sel):
                return False
        return True

    def _extract_description(self, soup) -> str | None:
        el = soup.select_one(
            "[itemprop='description'], "
            ".product-description"
        )
        return el.text.strip()[:500] if el else None

    def _extract_image(self, soup) -> str | None:
        el = soup.select_one(
            "[itemprop='image'], "
            ".product-image img"
        )
        return el.get("src") if el else None

    def _extract_currency(self, soup) -> str:
        el = soup.select_one(
            "[itemprop='priceCurrency']"
        )
        return el.get("content", "USD") if el else "USD"

    def _get_domain(self, url: str) -> str:
        from urllib.parse import urlparse
        return urlparse(url).netloc

Step 4: Validation with Pydantic

Validate parsed data before loading:

# validator.py
from pydantic import ValidationError
from models import Product
from typing import Generator

class DataValidator:
    def __init__(self):
        self.valid_count = 0
        self.invalid_count = 0
        self.errors = []

    def validate(
        self, raw_data: dict
    ) -> Product | None:
        try:
            product = Product(**raw_data)
            self.valid_count += 1
            return product
        except ValidationError as e:
            self.invalid_count += 1
            self.errors.append({
                "url": raw_data.get("url", "unknown"),
                "errors": e.errors(),
            })
            return None

    def validate_batch(
        self, items: list[dict]
    ) -> Generator[Product, None, None]:
        for item in items:
            product = self.validate(item)
            if product:
                yield product

    @property    def stats(self) -> dict:
        total = self.valid_count + self.invalid_count
        return {
            "valid": self.valid_count,
            "invalid": self.invalid_count,
            "rate": (
                f"{self.valid_count/total:.1%}" 
                if total > 0 else "N/A"
            ),
        }

Step 5: Deduplication

Don't insert the same product twice:

# deduper.py
import hashlib
from typing import Generator

class Deduplicator:
    """Simple set-based deduplication.
    For production, use Redis or a bloom filter."""

    def __init__(self):
        self.seen = set()
        self.dupes_count = 0

    def _make_key(self, product) -> str:
        """Create a unique key for each product."""
        raw = f"{product.source_domain}:{product.title}:{product.price}"
        return hashlib.md5(raw.encode()).hexdigest()

    def is_duplicate(self, product) -> bool:
        key = self._make_key(product)
        if key in self.seen:
            self.dupes_count += 1
            return True
        self.seen.add(key)
        return False

    def filter_batch(
        self, products
    ) -> Generator:
        for product in products:
            if not self.is_duplicate(product):
                yield product

Step 6: Database Loader

# loader.py
import asyncpg
from models import Product

class PostgresLoader:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool = None
        self.loaded_count = 0

    async def connect(self):
        self.pool = await asyncpg.create_pool(
            self.dsn, min_size=2, max_size=10
        )
        await self._create_table()

    async def _create_table(self):
        async with self.pool.acquire() as conn:
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS products (
                    id SERIAL PRIMARY KEY,
                    url TEXT UNIQUE,
                    title TEXT NOT NULL,
                    price DECIMAL(10,2),
                    currency VARCHAR(3),
                    in_stock BOOLEAN,
                    description TEXT,
                    image_url TEXT,
                    scraped_at TIMESTAMP,
                    source_domain TEXT,
                    created_at TIMESTAMP DEFAULT NOW(),
                    updated_at TIMESTAMP DEFAULT NOW()
                )
            """)

    async def load(self, product: Product):
        async with self.pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO products 
                    (url, title, price, currency, 
                     in_stock, description, image_url,
                     scraped_at, source_domain)
                VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
                ON CONFLICT (url) DO UPDATE SET
                    price = EXCLUDED.price,
                    in_stock = EXCLUDED.in_stock,
                    scraped_at = EXCLUDED.scraped_at,
                    updated_at = NOW()
            """,
                str(product.url),
                product.title,
                product.price,
                product.currency,
                product.in_stock,
                product.description,
                str(product.image_url) if product.image_url else None,
                product.scraped_at,
                product.source_domain,
            )
            self.loaded_count += 1

    async def load_batch(self, products: list[Product]):
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                for product in products:
                    await self.load(product)

    async def close(self):
        if self.pool:
            await self.pool.close()

Step 7: The Pipeline Orchestrator

Wire everything together:

# pipeline.py
import asyncio
import aiohttp
import time
from fetcher import Fetcher
from parser import ProductParser
from validator import DataValidator
from deduper import Deduplicator
from loader import PostgresLoader

class ScrapingPipeline:
    def __init__(
        self,
        urls: list[str],
        db_dsn: str,
        max_concurrent: int = 20,
    ):
        self.urls = urls
        self.fetcher = Fetcher(
            max_concurrent=max_concurrent,
            captcha_solver=CaptchaSolver(
                api_base="https://www.passxapi.com"
            ),
        )
        self.parser = ProductParser()
        self.validator = DataValidator()
        self.deduper = Deduplicator()
        self.loader = PostgresLoader(db_dsn)

    async def run(self):
        start = time.monotonic()
        print(f"Starting pipeline for {len(self.urls)} URLs")

        # Connect to database
        await self.loader.connect()

        async with aiohttp.ClientSession(
            headers={
                "User-Agent": (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36"
                )
            }
        ) as session:
            # Process in batches
            batch_size = 50
            for i in range(0, len(self.urls), batch_size):
                batch = self.urls[i:i + batch_size]
                await self._process_batch(session, batch)

                # Progress report
                elapsed = time.monotonic() - start
                print(
                    f"  Batch {i//batch_size + 1}: "
                    f"{self.fetcher.stats} | "
                    f"Valid: {self.validator.stats} | "
                    f"Loaded: {self.loader.loaded_count} | "
                    f"Dupes: {self.deduper.dupes_count} | "
                    f"{elapsed:.0f}s elapsed"
                )

        await self.loader.close()

        elapsed = time.monotonic() - start
        self._print_summary(elapsed)

    async def _process_batch(
        self, session, urls: list[str]
    ):
        # Extract: fetch all URLs concurrently
        fetch_tasks = [
            self.fetcher.fetch(session, url) 
            for url in urls
        ]
        results = await asyncio.gather(*fetch_tasks)

        # Transform: parse + validate + dedupe
        products = []
        for result in results:
            if result.error or not result.html:
                continue

            # Parse
            raw = self.parser.parse(result.html, result.url)
            if not raw:
                continue

            # Validate
            product = self.validator.validate(raw)
            if not product:
                continue

            # Deduplicate
            if not self.deduper.is_duplicate(product):
                products.append(product)

        # Load: batch insert
        if products:
            await self.loader.load_batch(products)

    def _print_summary(self, elapsed: float):
        print(f"\n{'='*50}")
        print(f"Pipeline completed in {elapsed:.0f}s")
        print(f"  Fetched: {self.fetcher.stats}")
        print(f"  Validation: {self.validator.stats}")
        print(f"  Duplicates: {self.deduper.dupes_count}")
        print(f"  Loaded to DB: {self.loader.loaded_count}")
        print(f"  Rate: {len(self.urls)/elapsed:.0f} URLs/s")
        print(f"{'='*50}")


# Run it
async def main():
    urls = load_urls_from_file("urls.txt")

    pipeline = ScrapingPipeline(
        urls=urls,
        db_dsn="postgresql://user:pass@localhost/scraping",        max_concurrent=20,
    )

    await pipeline.run()

asyncio.run(main())

Adding a Dead Letter Queue

Pages that fail should be retried later, not lost:

# dlq.py
import json
from pathlib import Path
from datetime import datetime

class DeadLetterQueue:
    def __init__(self, path: str = "failed_urls.jsonl"):
        self.path = Path(path)
        self.count = 0

    def add(
        self, url: str, 
        error: str, 
        attempt: int = 1
    ):
        entry = {
            "url": url,
            "error": error,
            "attempt": attempt,
            "timestamp": datetime.utcnow().isoformat(),
        }
        with open(self.path, "a") as f:
            f.write(json.dumps(entry) + "\n")
        self.count += 1

    def get_retryable(
        self, max_attempts: int = 3
    ) -> list[str]:
        if not self.path.exists():
            return []

        retryable = []
        with open(self.path) as f:
            for line in f:
                entry = json.loads(line)
                if entry["attempt"] < max_attempts:
                    retryable.append(entry["url"])

        return retryable

Key Takeaways

Model your output first — define Pydantic models before writing parsers
CAPTCHAs are just a pipeline step — detect, solve, continue
Validate everything — bad data is worse than no data
Deduplicate before loading — saves DB writes and storage
Batch operations — DB inserts, HTTP requests, everything works better in batches
Track stats at every stage — know where your data is being lost
Dead letter queue — never lose a URL, retry later

For the CAPTCHA-solving step in your pipeline, check out passxapi-python — it provides async and sync clients that integrate cleanly into ETL workflows.

How do you structure your scraping pipelines? What stages do you find most important? Let me know in the comments.

Bypassing Cloudflare's Challenge Pages: JS Challenge, Managed Challenge, and Turnstile Explained

Alex Chen — Mon, 23 Mar 2026 00:59:59 +0000

You make a request to a website and instead of the page you wanted, you get a "Checking your browser..." screen. Or worse, a CAPTCHA. Welcome to Cloudflare's challenge system.

Cloudflare doesn't just use one type of protection — it has multiple challenge layers, and understanding which one you're facing is the first step to handling it.

The Three Challenge Types

1. JS Challenge (Non-Interactive)

The "Checking your browser..." page. No user interaction needed — Cloudflare runs JavaScript checks and redirects automatically if your browser passes.

What it checks:

JavaScript execution capability
Browser fingerprint consistency
TLS fingerprint (JA3/JA4)
IP reputation

Request → CF Edge → JS Challenge Page → 
  Browser runs checks (2-5s) → 
  Sets cf_clearance cookie → 
  Redirect to actual page

2. Managed Challenge (Adaptive)

Cloudflare decides at the edge whether to:

Pass you through silently (low risk)
Show a Turnstile widget (medium risk)
Show an interactive CAPTCHA (high risk)

This is the most common challenge type in 2024+. The decision happens server-side based on your risk score.

3. Turnstile (CAPTCHA Widget)

A standalone CAPTCHA widget that sites embed in their forms. Unlike the other two, Turnstile is explicitly placed by the site developer — it's not auto-injected by Cloudflare.

Identifying Which Challenge You're Facing

import httpx
from urllib.parse import urlparse

def identify_cf_challenge(resp: httpx.Response) -> str:
    """Identify the type of Cloudflare challenge."""

    # Check status code
    if resp.status_code == 403:
        if "cf-mitigated" in resp.headers.get(
            "cf-mitigated", ""
        ):
            return "blocked"  # IP banned

    if resp.status_code == 503:
        body = resp.text

        if "jschl-answer" in body or "challenge-platform" in body:
            return "js_challenge"

        if "managed-challenge" in body:
            return "managed_challenge"

    # Check for Turnstile in normal pages
    if resp.status_code == 200:
        body = resp.text
        if "cf-turnstile" in body or "challenges.cloudflare.com/turnstile" in body:
            return "turnstile"

    # Check headers
    if "cf-ray" in resp.headers:
        if resp.status_code in (403, 503):
            return "cf_block"

    return "none"

Handling JS Challenges

JS challenges require a real browser — you can't solve them with plain HTTP requests because they need JavaScript execution.

from playwright.sync_api import sync_playwright
from playwright_stealth import stealth_sync
import time

class CloudflareBypass:
    def __init__(self):
        self.cookies = {}
        self.user_agent = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/121.0.0.0 Safari/537.36"
        )

    def solve_js_challenge(self, url: str) -> dict:
        """Get cf_clearance cookie by solving 
        JS challenge in a real browser."""

        with sync_playwright() as p:
            browser = p.chromium.launch(
                headless=False,  # Headed works better
                args=[
                    "--disable-blink-features="
                    "AutomationControlled"
                ]
            )

            context = browser.new_context(
                user_agent=self.user_agent,
                viewport={"width": 1920, "height": 1080},
                locale="en-US",
            )

            page = context.new_page()
            stealth_sync(page)

            # Navigate and wait for challenge to complete
            page.goto(url)

            # Wait for cf_clearance cookie to appear
            for _ in range(30):
                cookies = context.cookies()
                cf_cookie = next(
                    (c for c in cookies 
                     if c["name"] == "cf_clearance"),
                    None
                )
                if cf_cookie:
                    # Challenge solved!
                    self.cookies = {
                        c["name"]: c["value"] 
                        for c in cookies
                    }
                    browser.close()
                    return self.cookies

                time.sleep(1)

            browser.close()
            raise TimeoutError(
                "JS challenge not solved in 30s"
            )

    def make_request(self, url: str) -> httpx.Response:
        """Make request using saved cookies."""
        return httpx.get(
            url,
            cookies=self.cookies,
            headers={"User-Agent": self.user_agent}
        )

Cookie Persistence

The cf_clearance\ cookie is valid for a configurable period (usually 15-30 minutes). Save and reuse it:

import json
from pathlib import Path

class CookieStore:
    def __init__(self, path: str = "cookies.json"):
        self.path = Path(path)

    def save(self, domain: str, cookies: dict):
        store = self.load_all()
        store[domain] = {
            "cookies": cookies,
            "saved_at": time.time(),
            "ttl": 1800,  # 30 minutes
        }
        self.path.write_text(json.dumps(store))

    def get(self, domain: str) -> dict | None:
        store = self.load_all()
        entry = store.get(domain)
        if not entry:
            return None

        # Check if expired
        elapsed = time.time() - entry["saved_at"]
        if elapsed > entry["ttl"]:
            return None

        return entry["cookies"]

    def load_all(self) -> dict:
        if self.path.exists():
            return json.loads(self.path.read_text())
        return {}

Handling Managed Challenges

Managed challenges adapt based on your risk score. The best strategy is a tiered approach:

class ManagedChallengeHandler:
    def __init__(self):
        self.cookie_store = CookieStore()
        self.cf_bypass = CloudflareBypass()
        self.captcha_solver = CaptchaSolver()

    async def handle(self, url: str) -> str:
        domain = urlparse(url).netloc

        # Tier 1: Try with saved cookies
        cookies = self.cookie_store.get(domain)
        if cookies:
            resp = httpx.get(
                url, cookies=cookies,
                headers={
                    "User-Agent": self.cf_bypass.user_agent
                }
            )
            if resp.status_code == 200:
                return resp.text

        # Tier 2: Solve via browser 
        # (handles JS + Managed silently)
        try:
            cookies = self.cf_bypass.solve_js_challenge(url)
            self.cookie_store.save(domain, cookies)
            resp = httpx.get(
                url, cookies=cookies,
                headers={
                    "User-Agent": self.cf_bypass.user_agent
                }
            )
            if resp.status_code == 200:
                return resp.text
        except TimeoutError:
            pass

        # Tier 3: If Turnstile appears, solve via API
        return await self.solve_turnstile_flow(url)

    async def solve_turnstile_flow(self, url: str):
        """Handle explicit Turnstile challenge."""
        # Get the sitekey
        resp = httpx.get(url)
        sitekey = extract_turnstile_sitekey(resp.text)

        if not sitekey:
            raise Exception(
                "Could not find Turnstile sitekey"
            )

        # Solve via API
        token = await self.captcha_solver.solve(
            captcha_type="turnstile",
            sitekey=sitekey,
            url=url
        )

        # Submit the token
        resp = httpx.post(url, data={
            "cf-turnstile-response": token
        })

        return resp.text

Handling Turnstile (Form-Embedded)

Turnstile in forms is the simplest to handle — it's just a CAPTCHA widget:

import httpx
import re

async def solve_turnstile_form(
    url: str, 
    form_data: dict
) -> httpx.Response:
    """Submit a form that has Turnstile protection."""

    # 1. Load the page to get the sitekey
    client = httpx.Client()
    page = client.get(url)

    # 2. Extract Turnstile sitekey
    match = re.search(
        r'class="cf-turnstile"[^>]*data-sitekey="([^"]+)"',
        page.text
    )
    if not match:
        # Try alternate patterns
        match = re.search(
            r"turnstile\.render\([^,]+,\s*\{[^}]*sitekey:\s*['\"]([^'\"]+)",
            page.text
        )

    sitekey = match.group(1) if match else None

    if not sitekey:
        raise ValueError("No Turnstile sitekey found")

    # 3. Solve
    solver = CaptchaSolver(
        api_base="https://www.passxapi.com"
    )
    token = await solver.solve(
        captcha_type="turnstile",
        sitekey=sitekey,
        url=url
    )

    # 4. Submit form with token
    form_data["cf-turnstile-response"] = token
    resp = client.post(url, data=form_data)

    return resp

TLS Fingerprinting: The Hidden Challenge

Even with cookies, Cloudflare checks your TLS fingerprint (JA3/JA4). Python's default TLS stack looks different from Chrome:

# Problem: Python's httpx/requests uses its own 
# TLS fingerprint that doesn't match Chrome

# Solution 1: Use curl_cffi (mimics browser TLS)
from curl_cffi import requests as cfreq

resp = cfreq.get(
    "https://target-site.com",
    impersonate="chrome120",  # Mimics Chrome 120
    cookies=cookies
)

# Solution 2: Use tls-client
import tls_client

session = tls_client.Session(
    client_identifier="chrome_120",
    random_tls_extension_order=True
)
resp = session.get(
    "https://target-site.com",
    cookies=cookies
)

The Complete Flow

import asyncio
from dataclasses import dataclass

@dataclassclass ScrapeResult:
    url: str
    html: str
    challenge_type: str
    cookies_reused: bool

class CloudflareScraper:
    def __init__(self):
        self.handler = ManagedChallengeHandler()
        self.stats = {
            "no_challenge": 0,
            "cookie_reuse": 0,
            "js_solved": 0,
            "turnstile_solved": 0,
            "failed": 0,
        }

    async def scrape(self, url: str) -> ScrapeResult:
        domain = urlparse(url).netloc

        # Try direct request first
        resp = httpx.get(url, follow_redirects=True)
        challenge = identify_cf_challenge(resp)

        if challenge == "none":
            self.stats["no_challenge"] += 1
            return ScrapeResult(
                url=url, html=resp.text,
                challenge_type="none",
                cookies_reused=False
            )

        # Handle the challenge
        try:
            html = await self.handler.handle(url)
            self.stats[f"{challenge}_solved"] = (
                self.stats.get(f"{challenge}_solved", 0) + 1
            )
            return ScrapeResult(
                url=url, html=html,
                challenge_type=challenge,
                cookies_reused=False
            )
        except Exception as e:
            self.stats["failed"] += 1
            raise

    async def scrape_many(
        self, urls: list[str]
    ) -> list[ScrapeResult]:
        results = []
        for url in urls:
            result = await self.scrape(url)
            results.append(result)

        print(f"Stats: {self.stats}")
        return results

Key Takeaways

Identify the challenge type first — JS, Managed, and Turnstile need different approaches
JS challenges need a real browser — but you can reuse the cookies afterward
Managed challenges are adaptive — lower your risk score with residential IPs and real fingerprints
TLS fingerprint matters — use curl_cffi or tls-client to match browser signatures
Cache cf_clearance cookies — they're valid for 15-30 minutes, don't re-solve every request
Turnstile is the easiest — it's just a CAPTCHA widget you can solve via API

For solving Turnstile and other CAPTCHAs when they appear, check out passxapi-python — it handles Turnstile, reCAPTCHA, hCaptcha, and FunCaptcha with a single API.

Dealing with a specific Cloudflare challenge setup? Share your approach in the comments.

Browser Fingerprinting and CAPTCHAs: Why Headless Chrome Gets Caught and How to Fix It

Alex Chen — Mon, 23 Mar 2026 00:53:13 +0000

You launch your Playwright scraper in headless mode. It works on page 1. By page 3, you're staring at a CAPTCHA. Switch to headed mode — no CAPTCHA for 50 pages.

What's going on? Browser fingerprinting. Anti-bot systems don't just check your IP — they analyze dozens of browser properties to decide if you're human. Let's break down exactly what they check and how to handle it.

What Is Browser Fingerprinting?

Every browser exposes properties through JavaScript APIs. Anti-bot services collect these into a "fingerprint" — a unique identifier for your browser session:

// Just a sample of what gets collected
{
  userAgent: navigator.userAgent,
  platform: navigator.platform,
  languages: navigator.languages,
  hardwareConcurrency: navigator.hardwareConcurrency,
  deviceMemory: navigator.deviceMemory,
  screenResolution: [screen.width, screen.height],
  colorDepth: screen.colorDepth,
  timezone: Intl.DateTimeFormat().resolvedOptions().timeZone,
  webglVendor: getWebGLVendor(),
  webglRenderer: getWebGLRenderer(),
  canvas: getCanvasFingerprint(),
  audioContext: getAudioFingerprint(),
  fonts: getInstalledFonts(),
  plugins: navigator.plugins.length,
  touchSupport: navigator.maxTouchPoints,
}

How Headless Chrome Gets Detected

Headless Chrome has several telltale signs:

1. The navigator.webdriver Flag

// Headless Chrome:
navigator.webdriver  // true ← BUSTED

// Real Chrome:
navigator.webdriver  // undefined or false

2. Missing Plugins

// Real Chrome: has PDF viewer, etc.
navigator.plugins.length  // 3-5

// Headless Chrome:
navigator.plugins.length  // 0 ← suspicious

3. WebGL Renderer

// Real Chrome:
getWebGLRenderer()  // "ANGLE (NVIDIA GeForce...)"

// Headless Chrome:
getWebGLRenderer()  // "Google SwiftShader" ← dead giveaway

4. Chrome Object

// Real Chrome:
window.chrome  // {runtime: {...}, ...}

// Headless Chrome:
window.chrome  // undefined ← missing

5. Permissions API Behavior

// Real Chrome:
navigator.permissions.query({name: "notifications"})
  .then(p => p.state)  // "prompt" or "denied"

// Headless Chrome sometimes:
// Throws or returns unexpected values

Detecting These Leaks in Your Scraper

Before trying to fix things, find out what's leaking:

# fingerprint_audit.py
from playwright.sync_api import sync_playwright

DETECTION_SCRIPT = """
() => {
    const results = {};

    // Test 1: webdriver flag
    results.webdriver = navigator.webdriver;

    // Test 2: plugins
    results.pluginCount = navigator.plugins.length;

    // Test 3: languages
    results.languages = navigator.languages;

    // Test 4: chrome object
    results.hasChrome = !!window.chrome;
    results.hasChromeRuntime = !!(
        window.chrome && window.chrome.runtime
    );

    // Test 5: WebGL
    try {
        const canvas = document.createElement('canvas');
        const gl = canvas.getContext('webgl');
        const debugInfo = gl.getExtension(
            'WEBGL_debug_renderer_info'
        );
        results.webglVendor = gl.getParameter(
            debugInfo.UNMASKED_VENDOR_WEBGL
        );
        results.webglRenderer = gl.getParameter(
            debugInfo.UNMASKED_RENDERER_WEBGL
        );
    } catch(e) {
        results.webglError = e.message;
    }

    // Test 6: Permissions
    results.permissionsAPI = !!navigator.permissions;

    // Test 7: Screen dimensions
    results.screen = {
        width: screen.width,
        height: screen.height,
        availWidth: screen.availWidth,
        availHeight: screen.availHeight,
        colorDepth: screen.colorDepth,
    };

    // Test 8: Hardware
    results.hardwareConcurrency = navigator.hardwareConcurrency;
    results.deviceMemory = navigator.deviceMemory;
    results.maxTouchPoints = navigator.maxTouchPoints;

    // Test 9: Headless indicators
    results.userAgent = navigator.userAgent;
    results.platform = navigator.platform;

    return results;
}
"""

def audit_fingerprint():
    with sync_playwright() as p:
        # Test headless
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        headless_fp = page.evaluate(DETECTION_SCRIPT)
        browser.close()

        # Compare with headed
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()
        headed_fp = page.evaluate(DETECTION_SCRIPT)
        browser.close()

        # Show differences
        print("=== Fingerprint Differences ===")
        for key in headless_fp:
            if headless_fp[key] != headed_fp.get(key):
                print(f"  {key}:")
                print(f"    Headless: {headless_fp[key]}")
                print(f"    Headed:   {headed_fp.get(key)}")

audit_fingerprint()

Fixing the Fingerprint Leaks

Approach 1: Playwright Stealth (Quick Fix)

Use playwright-stealth\ to patch common detection points:

# pip install playwright-stealth
from playwright.sync_api import sync_playwright
from playwright_stealth import stealth_sync

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()

    # Apply stealth patches
    stealth_sync(page)

    page.goto("https://target-site.com")
    # Now navigator.webdriver = false,
    # plugins are spoofed, etc.

Approach 2: Manual Patches (More Control)

from playwright.sync_api import sync_playwright

def apply_stealth(page):
    """Apply individual stealth patches."""

    # 1. Remove webdriver flag
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });
    """)

    # 2. Fake plugins
    page.add_init_script("""
        Object.defineProperty(navigator, 'plugins', {
            get: () => {
                const plugins = [
                    {
                        name: 'Chrome PDF Plugin',
                        description: 'Portable Document Format',
                        filename: 'internal-pdf-viewer',
                        length: 1
                    },
                    {
                        name: 'Chrome PDF Viewer',
                        description: '',
                        filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai',
                        length: 1
                    },
                    {
                        name: 'Native Client',
                        description: '',
                        filename: 'internal-nacl-plugin',
                        length: 2
                    }
                ];
                plugins.length = 3;
                return plugins;
            }
        });
    """)

    # 3. Fake chrome object
    page.add_init_script("""
        window.chrome = {
            runtime: {
                onConnect: null,
                onMessage: null,
            },
            loadTimes: function() {},
            csi: function() {},
            app: {}
        };
    """)

    # 4. Fix permissions query
    page.add_init_script("""
        const originalQuery = window.navigator.permissions.query;
        window.navigator.permissions.query = (params) => {
            if (params.name === 'notifications') {
                return Promise.resolve({
                    state: Notification.permission
                });
            }
            return originalQuery(params);
        };
    """)

    # 5. Fake languages
    page.add_init_script("""
        Object.defineProperty(navigator, 'languages', {
            get: () => ['en-US', 'en']
        });
    """)

with sync_playwright() as p:
    browser = p.chromium.launch(
        headless=True,
        args=[
            '--disable-blink-features=AutomationControlled',
        ]
    )

    context = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent=(
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/120.0.0.0 Safari/537.36"
        ),
        locale="en-US",
        timezone_id="America/New_York",
    )

    page = context.new_page()
    apply_stealth(page)
    page.goto("https://target-site.com")

Approach 3: CDP Connection to Real Browser

The most reliable approach — connect to a real browser instance:

from playwright.sync_api import sync_playwright

# Launch a real Chrome with remote debugging
# chrome --remote-debugging-port=9222

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(
        "http://localhost:9222"
    )

    context = browser.contexts[0]
    page = context.new_page()

    # This IS a real browser — nothing to patch
    page.goto("https://target-site.com")

When Stealth Isn't Enough: Solving CAPTCHAs

Even with perfect fingerprinting, some sites will still show CAPTCHAs — especially on:

First visit from a new IP
Login/signup flows
After N requests in a session
High-value pages (checkout, pricing)

That's when you need a solving service:

from playwright.sync_api import sync_playwright
from playwright_stealth import stealth_sync
import httpx

def scrape_with_stealth_and_solving(url: str):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        stealth_sync(page)

        page.goto(url)

        # Check if we got a CAPTCHA despite stealth
        captcha_el = page.query_selector(
            '[class*="captcha"], '
            '[data-sitekey], '
            '.g-recaptcha, '
            '.h-captcha, '
            '.cf-turnstile'
        )

        if captcha_el:
            print("CAPTCHA detected despite stealth — solving...")
            sitekey = (
                captcha_el.get_attribute("data-sitekey") 
                or extract_sitekey(page.content())
            )
            captcha_type = detect_type(captcha_el)

            # Solve via API
            token = solve_captcha(
                captcha_type=captcha_type,
                sitekey=sitekey,
                url=url
            )

            # Inject token
            page.evaluate(f"""() => {{
                const textarea = document.querySelector(
                    'textarea[name*="captcha-response"]'
                );
                if (textarea) textarea.value = '{token}';
            }}""")

            page.click('button[type="submit"]')
            page.wait_for_load_state("networkidle")

        # Now scrape the actual content
        data = extract_data(page)
        browser.close()
        return data

Fingerprint Consistency Checklist

When setting up your scraper, make sure these are all consistent:

# ✅ Consistent setup
context = browser.new_context(
    # Match the User-Agent to the viewport/platform
    user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
    viewport={"width": 1920, "height": 1080},  # Desktop
    locale="en-US",
    timezone_id="America/New_York",

    # Screen size should match viewport
    screen={"width": 1920, "height": 1080},

    # Match color scheme to majority of users
    color_scheme="light",
)

# ❌ Inconsistent (will get flagged)
context = browser.new_context(
    user_agent="...iPhone...",       # Says mobile
    viewport={"width": 1920, ...},  # But desktop viewport!
    locale="zh-CN",                 # Chinese locale
    timezone_id="America/New_York", # But US timezone!
)

Testing Your Stealth Setup

Run your browser against common detection sites:

DETECTION_SITES = [
    "https://bot.sannysoft.com/",
    "https://arh.antoinevastel.com/bots/areyouheadless",
    "https://infosimples.github.io/detect-headless/",
    "https://fingerprintjs.github.io/fingerprintjs/",
]

def test_stealth(page):
    for site in DETECTION_SITES:
        page.goto(site)
        page.wait_for_timeout(3000)
        page.screenshot(
            path=f"stealth_test_{site.split('/')[2]}.png"
        )
        print(f"Screenshotted: {site}")

The Decision Tree

Start
  ├─ Can you use a real browser (CDP)?
  │   └─ Yes → Best stealth, highest resource cost
  ├─ Is the site moderately protected?
  │   └─ playwright-stealth + good fingerprint config
  ├─ Is the site heavily protected?
  │   └─ Stealth + CAPTCHA solving API for fallback
  └─ Is it an API endpoint (no JS)?
      └─ Skip browser entirely, just solve CAPTCHAs
          via HTTP and submit tokens

Key Takeaways

Headless detection is fingerprint-based — not just User-Agent
WebGL renderer and navigator.webdriver are the biggest tells
Consistency matters — a mobile UA with a desktop viewport is obvious
Stealth plugins fix 80% — but determined anti-bot systems need more
Always have a CAPTCHA solver fallback — perfect stealth is impossible

For solving CAPTCHAs when stealth isn't enough, check out passxapi-python — it handles reCAPTCHA, hCaptcha, Turnstile, and FunCaptcha with a unified API.

What's your stealth setup? Have you found detection vectors I didn't cover? Let me know in the comments.

Adding CAPTCHA Protection to Your Django or Flask App (And How Attackers Bypass It)

Alex Chen — Mon, 23 Mar 2026 00:45:59 +0000

You've built a web app. Users are signing up, submitting forms, leaving comments. Then the bots arrive — fake signups, spam comments, credential stuffing.

You need CAPTCHA protection. But adding it wrong is worse than not adding it at all — it gives you a false sense of security while bots walk right through.

This guide shows you how to properly integrate CAPTCHAs in Django and Flask, the common mistakes that make them bypassable, and how to verify you've done it right.

Choosing a CAPTCHA Provider

Provider	Difficulty	Privacy	Free Tier	Best For
reCAPTCHA v2	Medium	Low (Google tracking)	10K/month	Legacy apps
reCAPTCHA v3	Invisible	Low	10K/month	UX-sensitive apps
hCaptcha	Medium	High	Unlimited	Privacy-conscious
Turnstile	Invisible	Medium	Unlimited	Cloudflare users

For this tutorial, we'll use hCaptcha — it's free, privacy-friendly, and works everywhere.

Django Integration

Step 1: Get Your Keys

Site Key (public, goes in HTML)
Secret Key (private, stays on server)

Step 2: Create a CAPTCHA Form Mixin

# captcha/mixins.py
import httpx
from django import forms
from django.conf import settings

class CaptchaMixin:
    """Add hCaptcha validation to any Django form."""

    def clean(self):
        cleaned_data = super().clean()

        captcha_response = self.data.get(
            "h-captcha-response", ""
        )

        if not captcha_response:
            raise forms.ValidationError(
                "Please complete the CAPTCHA."
            )

        # Verify server-side — NEVER skip this
        result = httpx.post(
            "https://hcaptcha.com/siteverify",
            data={
                "secret": settings.HCAPTCHA_SECRET_KEY,
                "response": captcha_response,
                "remoteip": self.get_client_ip(),
            }
        ).json()

        if not result.get("success"):
            raise forms.ValidationError(
                "CAPTCHA verification failed. "
                "Please try again."
            )

        return cleaned_data

    def get_client_ip(self):
        """Extract client IP from the request."""
        x_forwarded_for = self.request.META.get(
            "HTTP_X_FORWARDED_FOR"
        )
        if x_forwarded_for:
            return x_forwarded_for.split(",")[0].strip()
        return self.request.META.get("REMOTE_ADDR")

Step 3: Apply to Your Forms

# accounts/forms.py
from django import forms
from captcha.mixins import CaptchaMixin

class SignupForm(CaptchaMixin, forms.Form):
    username = forms.CharField(max_length=150)
    email = forms.EmailField()
    password = forms.CharField(
        widget=forms.PasswordInput
    )

    def __init__(self, *args, request=None, **kwargs):
        self.request = request
        super().__init__(*args, **kwargs)

class ContactForm(CaptchaMixin, forms.Form):
    name = forms.CharField(max_length=100)
    message = forms.CharField(
        widget=forms.Textarea
    )

    def __init__(self, *args, request=None, **kwargs):
        self.request = request
        super().__init__(*args, **kwargs)

Step 4: Template

<!-- templates/signup.html -->
<form method="post">
    {% csrf_token %}
    {{ form.as_p }}

    <!-- hCaptcha widget -->
    <div class="h-captcha" 
         data-sitekey="{{ HCAPTCHA_SITE_KEY }}">
    </div>

    <button type="submit">Sign Up</button>
</form>

<script src="https://js.hcaptcha.com/1/api.js" 
        async defer></script>

Step 5: View

# accounts/views.py
from django.shortcuts import render, redirect
from django.conf import settings
from .forms import SignupForm

def signup(request):
    if request.method == "POST":
        form = SignupForm(
            request.POST, request=request
        )
        if form.is_valid():
            # CAPTCHA passed — create user
            create_user(form.cleaned_data)
            return redirect("dashboard")
    else:
        form = SignupForm()

    return render(request, "signup.html", {
        "form": form,
        "HCAPTCHA_SITE_KEY": settings.HCAPTCHA_SITE_KEY,
    })

Flask Integration

Step 1: CAPTCHA Decorator

# captcha.py
import httpx
import functools
from flask import request, abort, current_app

def require_captcha(f):
    """Decorator to require CAPTCHA verification."""
    @functools.wraps(f)    def decorated(*args, **kwargs):
        if request.method != "POST":
            return f(*args, **kwargs)

        token = request.form.get(
            "h-captcha-response", ""
        )

        if not token:
            abort(400, "Missing CAPTCHA response")

        result = httpx.post(
            "https://hcaptcha.com/siteverify",
            data={
                "secret": current_app.config[
                    "HCAPTCHA_SECRET"
                ],
                "response": token,
                "remoteip": request.remote_addr,
            }
        ).json()

        if not result.get("success"):
            abort(403, "CAPTCHA verification failed")

        return f(*args, **kwargs)

    return decorated

Step 2: Apply to Routes

# app.py
from flask import Flask, render_template, request
from captcha import require_captcha

app = Flask(__name__)
app.config["HCAPTCHA_SECRET"] = "your-secret-key"
app.config["HCAPTCHA_SITEKEY"] = "your-site-key"

@app.route("/signup", methods=["GET", "POST"])
@require_captchadef signup():
    if request.method == "POST":
        # CAPTCHA already verified by decorator
        username = request.form["username"]
        email = request.form["email"]
        create_user(username, email)
        return redirect("/dashboard")

    return render_template(
        "signup.html",
        sitekey=app.config["HCAPTCHA_SITEKEY"]
    )

@app.route("/contact", methods=["GET", "POST"])
@require_captchadef contact():
    if request.method == "POST":
        send_message(request.form["message"])
        return redirect("/thanks")

    return render_template(
        "contact.html",
        sitekey=app.config["HCAPTCHA_SITEKEY"]
    )

Common Mistakes (And How Attackers Exploit Them)

Mistake 1: Client-Side Only Validation

# BAD — no server-side check
@app.route("/signup", methods=["POST"])
def signup():
    # Just trusts the form submission...
    create_user(request.form["username"])

Attack: Simply POST to /signup\ without the CAPTCHA field. The server never checks.

Fix: Always verify the token server-side with the siteverify\ endpoint.

Mistake 2: Not Checking the Response

# BAD — sends verification but ignores result
result = httpx.post(
    "https://hcaptcha.com/siteverify",
    data={"secret": secret, "response": token}
).json()
# Forgot to check result["success"]!
create_user(data)

Attack: Send any random string as the CAPTCHA token. The server verifies it, gets success: false\, and ignores it.

Mistake 3: Reusable Tokens

# BAD — doesn't track used tokens
@app.route("/vote", methods=["POST"])
@require_captchadef vote():
    add_vote(request.form["option"])

Attack: Solve CAPTCHA once, replay the same token on multiple votes. hCaptcha tokens are single-use by default at the provider level, but you should also track them:

# GOOD — track used tokens
used_tokens = set()

def verify_captcha(token):
    if token in used_tokens:
        return False

    result = httpx.post(
        "https://hcaptcha.com/siteverify",
        data={"secret": secret, "response": token}
    ).json()

    if result["success"]:
        used_tokens.add(token)
        return True
    return False

Mistake 4: CAPTCHA on GET, Not POST

# BAD — CAPTCHA only shown on the form page
@app.route("/api/submit", methods=["POST"])
def api_submit():
    # No CAPTCHA check — assumes user went through
    # the form page first
    process(request.json)

Attack: Call the API endpoint directly, skipping the form page entirely.

How Scrapers Bypass Your CAPTCHA

Understanding the attacker's perspective helps you build better defenses. Here's how automated tools handle CAPTCHAs:

# What a scraper does to bypass your form
import httpx

# 1. Load the page, find the sitekey
resp = httpx.get("https://yourapp.com/signup")
sitekey = extract_sitekey(resp.text)

# 2. Solve the CAPTCHA via API
token = solve_captcha(
    type="hcaptcha", 
    sitekey=sitekey,
    url="https://yourapp.com/signup"
)

# 3. Submit the form with the solved token
httpx.post("https://yourapp.com/signup", data={
    "username": "bot_user",
    "email": "bot@example.com",    "h-captcha-response": token
})

Tools like passxapi-python make this straightforward. The solve produces a valid token that passes server-side verification.

Defense in Depth

CAPTCHAs alone aren't enough. Layer your defenses:

# middleware.py
from flask import request, abort
import time

class AntiAbuseMiddleware:
    def __init__(self, app):
        self.app = app
        self.rate_limiter = RateLimiter()
        self.fingerprinter = BrowserFingerprint()

    def __call__(self, environ, start_response):
        with self.app.request_context(environ):
            ip = request.remote_addr

            # Layer 1: Rate limiting
            if self.rate_limiter.is_exceeded(ip):
                abort(429)

            # Layer 2: Request timing analysis
            if self.is_too_fast(request):
                abort(429)

            # Layer 3: CAPTCHA (handled by decorator)
            # Layer 4: Email verification (post-signup)
            # Layer 5: Behavioral analysis

        return self.app(environ, start_response)

    def is_too_fast(self, req):
        """Flag requests submitted faster than 
        a human could type."""
        form_load_time = req.form.get("_form_loaded")
        if form_load_time:
            elapsed = time.time() - float(form_load_time)
            if elapsed < 3:  # < 3 seconds = suspicious
                return True
        return False

Add Hidden Honeypot Fields

<!-- Bots fill hidden fields, humans don't -->
<div style="display:none">
    <input type="text" 
           name="website" 
           tabindex="-1" 
           autocomplete="off">
</div>

@app.route("/signup", methods=["POST"])
@require_captchadef signup():
    # Honeypot check
    if request.form.get("website"):
        # Bot filled the hidden field
        return redirect("/thanks")  # Fake success

    create_user(request.form["username"])

Testing Your CAPTCHA Integration

Use test keys to verify without solving real CAPTCHAs:

# settings.py (test configuration)
import os

if os.getenv("TESTING"):
    # hCaptcha test keys — always pass
    HCAPTCHA_SITE_KEY = "10000000-ffff-ffff-ffff-000000000001"
    HCAPTCHA_SECRET_KEY = "0x0000000000000000000000000000000000000000"
else:
    HCAPTCHA_SITE_KEY = os.getenv("HCAPTCHA_SITE_KEY")
    HCAPTCHA_SECRET_KEY = os.getenv("HCAPTCHA_SECRET_KEY")

# tests/test_signup.py
import pytest

@pytest.fixturedef app():
    os.environ["TESTING"] = "1"
    from app import create_app
    return create_app()

def test_signup_with_captcha(client):
    resp = client.post("/signup", data={
        "username": "testuser",
        "email": "test@example.com",        "h-captcha-response": "10000000-aaaa-bbbb-cccc-000000000001"
    })
    assert resp.status_code == 302  # Redirect to dashboard

def test_signup_without_captcha(client):
    resp = client.post("/signup", data={
        "username": "testuser",
        "email": "test@example.com",        # No CAPTCHA token
    })
    assert resp.status_code == 400

Key Takeaways

Always verify server-side — client-side CAPTCHA is just UI
Check the success field — don't just send the request
CAPTCHAs are speed bumps, not walls — layer with rate limiting, honeypots, email verification
Use test keys in CI — don't skip CAPTCHA checks in tests
Understand the attacker — if you know how CAPTCHAs get solved, you can build better defenses

What CAPTCHA provider are you using in production? How do you layer your defenses? Let's discuss in the comments.

Why Your Scraper's Location Matters: Geo-Distributed Scraping with CAPTCHA Handling

Alex Chen — Mon, 23 Mar 2026 00:39:36 +0000

You built a scraper that works perfectly from your laptop in San Francisco. You deploy it to a server in Frankfurt and suddenly CAPTCHAs appear on every request.

What changed? Your geographic location. CAPTCHAs are location-aware, and understanding this can make or break your scraping infrastructure.

How Location Affects CAPTCHA Difficulty

Anti-bot systems consider your IP's geographic context:

Factor	Low Risk	High Risk
IP location vs site audience	Same country	Different continent
Datacenter vs residential	Residential	Known datacenter
IP reputation	Clean history	Previous abuse
Region consistency	Stable	Jumping between countries

A residential IP in the US scraping a US e-commerce site? Low CAPTCHA rate. A Hetzner IP in Germany scraping the same site? CAPTCHA on every page.

Measuring the Impact

Before optimizing, measure your baseline:

import httpx
import time
from dataclasses import dataclass

@dataclassclass ProbeResult:
    region: str
    url: str
    captcha_detected: bool
    response_time: float
    status_code: int

async def probe_from_region(
    proxy: str, 
    region: str, 
    target_url: str
) -> ProbeResult:
    """Test CAPTCHA rate from a specific region."""
    start = time.monotonic()

    async with httpx.AsyncClient(
        proxies={"all://": proxy}
    ) as client:
        resp = await client.get(target_url)
        html = resp.text

        captcha = any(keyword in html for keyword in [
            "g-recaptcha", "h-captcha", 
            "cf-turnstile", "arkoselabs",
            "challenge-platform"
        ])

        return ProbeResult(
            region=region,
            url=target_url,
            captcha_detected=captcha,
            response_time=time.monotonic() - start,
            status_code=resp.status_code
        )

# Test from multiple regions
proxies = {
    "us-east": "socks5://proxy-us-east:1080",
    "us-west": "socks5://proxy-us-west:1080",
    "eu-west": "socks5://proxy-eu-west:1080",
    "ap-south": "socks5://proxy-ap-south:1080",
}

results = []
for region, proxy in proxies.items():
    for _ in range(10):  # 10 probes per region
        result = await probe_from_region(
            proxy, region, "https://target-site.com"
        )
        results.append(result)

# Analyze
for region in proxies:
    region_results = [r for r in results if r.region == region]
    captcha_rate = sum(
        1 for r in region_results if r.captcha_detected
    ) / len(region_results)
    avg_time = sum(
        r.response_time for r in region_results
    ) / len(region_results)
    print(
        f"{region}: CAPTCHA rate={captcha_rate:.0%}, "
        f"avg={avg_time:.2f}s"
    )

Typical output:

us-east:  CAPTCHA rate=10%, avg=0.45s
us-west:  CAPTCHA rate=15%, avg=0.52s
eu-west:  CAPTCHA rate=60%, avg=0.89s
ap-south: CAPTCHA rate=85%, avg=1.23s

Architecture: Geo-Distributed Scraper

Option 1: Region-Aware Proxy Selection

Route requests through proxies close to the target site's audience:

from dataclasses import dataclass, field

@dataclassclass RegionProxy:
    region: str
    proxy_url: str
    captcha_rate: float = 0.0
    total_requests: int = 0
    captcha_hits: int = 0

    def update_stats(self, had_captcha: bool):
        self.total_requests += 1
        if had_captcha:
            self.captcha_hits += 1
        self.captcha_rate = (
            self.captcha_hits / self.total_requests
        )


class GeoRouter:
    """Route requests to the best region 
    based on CAPTCHA rates."""

    def __init__(self, proxies: list[RegionProxy]):
        self.proxies = proxies
        self._lock = asyncio.Lock()

    async def get_best_proxy(
        self, target_domain: str
    ) -> RegionProxy:
        async with self._lock:
            # Sort by CAPTCHA rate (lower is better)
            # But ensure minimum samples
            candidates = [
                p for p in self.proxies
                if p.total_requests >= 5
            ]

            if not candidates:
                # Not enough data — round robin
                return self.proxies[
                    hash(target_domain) % len(self.proxies)
                ]

            return min(
                candidates, key=lambda p: p.captcha_rate
            )

    async def report_result(
        self, proxy: RegionProxy, had_captcha: bool
    ):
        async with self._lock:
            proxy.update_stats(had_captcha)


# Setup
router = GeoRouter([
    RegionProxy("us-east", "socks5://us-east:1080"),
    RegionProxy("us-west", "socks5://us-west:1080"),
    RegionProxy("eu-west", "socks5://eu-west:1080"),
])

Option 2: Multi-Region Workers

Deploy workers in multiple regions and distribute work based on target site location:

import asyncio
from enum import Enum

class Region(Enum):
    US = "us"
    EU = "eu"
    ASIA = "asia"

# Map target domains to preferred regions
DOMAIN_REGION_MAP = {
    "amazon.com": Region.US,
    "amazon.co.uk": Region.EU,
    "amazon.co.jp": Region.ASIA,
    "ebay.com": Region.US,
    "zalando.de": Region.EU,
}

def get_preferred_region(url: str) -> Region:
    from urllib.parse import urlparse
    domain = urlparse(url).netloc

    # Check exact match
    if domain in DOMAIN_REGION_MAP:
        return DOMAIN_REGION_MAP[domain]

    # Check TLD
    tld = domain.split(".")[-1]
    tld_map = {
        "uk": Region.EU, "de": Region.EU,
        "fr": Region.EU, "it": Region.EU,
        "jp": Region.ASIA, "kr": Region.ASIA,
        "cn": Region.ASIA, "in": Region.ASIA,
    }
    return tld_map.get(tld, Region.US)


class DistributedScraper:
    def __init__(self):
        self.workers = {
            Region.US: WorkerPool("us-queue"),
            Region.EU: WorkerPool("eu-queue"),
            Region.ASIA: WorkerPool("asia-queue"),
        }

    async def scrape(self, url: str) -> dict:
        region = get_preferred_region(url)
        worker = self.workers[region]
        return await worker.submit(url)

Handling CAPTCHAs When They Do Appear

Even with the right region, some CAPTCHAs are inevitable. Handle them without blocking:

class GeoAwareScraper:
    def __init__(self, router: GeoRouter):
        self.router = router
        self.solver = CaptchaSolver(
            api_base="https://www.passxapi.com"
        )

    async def fetch(self, url: str) -> dict:
        proxy = await self.router.get_best_proxy(
            urlparse(url).netloc
        )

        async with httpx.AsyncClient(
            proxies={"all://": proxy.proxy_url}
        ) as client:
            resp = await client.get(url)
            html = resp.text

            captcha = detect_captcha(html)

            if captcha:
                # Report CAPTCHA hit to router
                await self.router.report_result(
                    proxy, had_captcha=True
                )

                # Solve it
                token = await self.solver.solve(
                    captcha_type=captcha["type"],
                    sitekey=captcha["sitekey"],
                    url=url
                )

                # Resubmit with token
                resp = await client.post(url, data={
                    captcha["field_name"]: token
                })
                html = resp.text
            else:
                await self.router.report_result(
                    proxy, had_captcha=False
                )

            return {
                "url": url, 
                "html": html,
                "region": proxy.region,
                "captcha_solved": captcha is not None
            }

Cost Optimization: Regional Pricing

Different proxy types have different costs. Optimize by mixing:

class CostOptimizedRouter:
    """Use cheap datacenter proxies when possible,
    expensive residential only when needed."""

    def __init__(self):
        self.tiers = {
            "datacenter": {
                "proxies": [...],
                "cost_per_gb": 0.50,
                "avg_captcha_rate": 0.40,
            },
            "residential": {
                "proxies": [...],
                "cost_per_gb": 8.00,
                "avg_captcha_rate": 0.05,
            },
        }

    def select_tier(
        self, target_url: str, 
        captcha_budget: float
    ) -> str:
        """Choose proxy tier based on 
        CAPTCHA cost tradeoff."""
        # Cost of CAPTCHA solve: ~$0.001
        # If datacenter CAPTCHA rate is 40%, 
        #   extra cost = 0.4 * $0.001 = $0.0004/req
        # Residential saves CAPTCHAs but costs more

        dc_total = (
            self.tiers["datacenter"]["cost_per_gb"] / 1000
            + self.tiers["datacenter"]["avg_captcha_rate"]
            * 0.001
        )
        res_total = (
            self.tiers["residential"]["cost_per_gb"] / 1000
            + self.tiers["residential"]["avg_captcha_rate"]
            * 0.001
        )

        # Use whichever is cheaper overall
        if dc_total < res_total:
            return "datacenter"
        return "residential"

Deployment: Docker Compose for Multi-Region

# docker-compose.yml
services:
  scraper-us:
    build: .
    environment:
      - REGION=us-east
      - PROXY_URL=socks5://us-proxy:1080
      - QUEUE_NAME=scrape-us
    deploy:
      replicas: 3

  scraper-eu:
    build: .
    environment:
      - REGION=eu-west
      - PROXY_URL=socks5://eu-proxy:1080
      - QUEUE_NAME=scrape-eu
    deploy:
      replicas: 2

  coordinator:
    build: .
    command: python coordinator.py
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  redis:
    image: redis:7-alpine

Key Takeaways

Measure first — probe CAPTCHA rates from different regions before choosing
Match your proxy location to the target audience — US site? Use US proxies
Adaptive routing — let your system learn which regions work best
Mix proxy tiers — datacenter for low-risk, residential for high-risk targets
Always have a CAPTCHA solver ready — even perfect geo-matching won't avoid all CAPTCHAs

For the CAPTCHA-solving integration used in these examples, check out passxapi-python — it works from any region with consistent solve rates.

What regions do you deploy your scrapers in? Share your geo-strategy in the comments.

Monitoring Your CAPTCHA Solver: Prometheus Metrics, Grafana Dashboards, and PagerDuty Alerts

Alex Chen — Mon, 23 Mar 2026 00:33:03 +0000

Your scraper runs fine for three days. On day four, CAPTCHA solve rates drop from 95% to 40%. You don't notice until a stakeholder asks why the data pipeline is empty.

Sound familiar? Let's build observability into your CAPTCHA-solving infrastructure so you catch problems before they catch you.

What to Monitor

Before writing any code, decide which metrics actually matter:

Metric	Why It Matters
Solve success rate	Primary health indicator
Solve latency (p50, p95, p99)	Catches degradation early
Token expiry rate	Are you submitting too slow?
Cost per solve	Budget tracking
Concurrent solves	Capacity planning
Error breakdown by type	Root cause analysis

Step 1: Instrument Your Solver

Wrap your CAPTCHA solver with Prometheus metrics:

# metrics.py
from prometheus_client import (
    Counter, Histogram, Gauge, Summary
)

# Counters
captcha_solves_total = Counter(
    "captcha_solves_total",
    "Total CAPTCHA solve attempts",
    ["captcha_type", "status"]  # status: success/failed/expired
)

captcha_errors_total = Counter(
    "captcha_errors_total",
    "CAPTCHA errors by type",
    ["captcha_type", "error_type"]
)

# Histograms for latency
captcha_solve_duration = Histogram(
    "captcha_solve_duration_seconds",
    "Time to solve a CAPTCHA",
    ["captcha_type"],
    buckets=[1, 2, 5, 10, 15, 20, 30, 45, 60, 90]
)

# Gauges for current state
captcha_concurrent_solves = Gauge(
    "captcha_concurrent_solves",
    "Currently in-progress CAPTCHA solves",
    ["captcha_type"]
)

captcha_cost_dollars = Counter(
    "captcha_cost_dollars_total",
    "Total cost of CAPTCHA solves in USD",
    ["captcha_type"]
)

# Token freshness
captcha_token_age_seconds = Histogram(
    "captcha_token_age_at_submission_seconds",
    "Age of token when submitted to target site",
    buckets=[1, 5, 10, 30, 60, 90, 120]
)

Step 2: Instrumented Solver Wrapper

# solver_instrumented.py
import time
import httpx
from contextlib import contextmanager
from metrics import (
    captcha_solves_total,
    captcha_solve_duration,
    captcha_concurrent_solves,
    captcha_errors_total,
    captcha_cost_dollars,
)

# Cost per solve by type
COST_PER_SOLVE = {
    "recaptcha_v2": 0.001,
    "recaptcha_v3": 0.001,
    "hcaptcha": 0.001,
    "turnstile": 0.001,
    "funcaptcha": 0.002,
}

class InstrumentedSolver:
    def __init__(
        self, 
        api_base: str = "https://www.passxapi.com"
    ):
        self.api_base = api_base
        self.client = httpx.Client(timeout=90)

    def solve(
        self, captcha_type: str, 
        sitekey: str, url: str
    ) -> str:
        # Track concurrent solves
        captcha_concurrent_solves.labels(
            captcha_type=captcha_type
        ).inc()

        start = time.monotonic()

        try:
            token = self._do_solve(
                captcha_type, sitekey, url
            )

            # Record success
            duration = time.monotonic() - start
            captcha_solves_total.labels(
                captcha_type=captcha_type,
                status="success"
            ).inc()
            captcha_solve_duration.labels(
                captcha_type=captcha_type
            ).observe(duration)
            captcha_cost_dollars.labels(
                captcha_type=captcha_type
            ).inc(COST_PER_SOLVE.get(captcha_type, 0.001))

            return token

        except TimeoutError:
            captcha_solves_total.labels(
                captcha_type=captcha_type,
                status="failed"
            ).inc()
            captcha_errors_total.labels(
                captcha_type=captcha_type,
                error_type="timeout"
            ).inc()
            raise

        except Exception as e:
            error_type = type(e).__name__.lower()
            captcha_solves_total.labels(
                captcha_type=captcha_type,
                status="failed"
            ).inc()
            captcha_errors_total.labels(
                captcha_type=captcha_type,
                error_type=error_type
            ).inc()
            raise

        finally:
            captcha_concurrent_solves.labels(
                captcha_type=captcha_type
            ).dec()

    def _do_solve(self, captcha_type, sitekey, url):
        resp = self.client.post(
            f"{self.api_base}/api/v1/task",
            json={
                "type": captcha_type,
                "sitekey": sitekey,
                "pageurl": url
            }
        )
        task_id = resp.json()["task_id"]

        for _ in range(60):
            result = self.client.get(
                f"{self.api_base}/api/v1/task/{task_id}"
            ).json()
            if result["status"] == "completed":
                return result["token"]
            if result["status"] == "failed":
                raise Exception(result.get("error"))
            time.sleep(2)

        raise TimeoutError("Solve timed out at 120s")

Step 3: Expose Metrics Endpoint

Add a /metrics\ endpoint to your scraper:

# server.py
from prometheus_client import (
    start_http_server, 
    generate_latest
)
import threading

def start_metrics_server(port: int = 9090):
    """Start Prometheus metrics server in background."""
    start_http_server(port)
    print(f"Metrics available at http://localhost:{port}/metrics")

# In your main scraper
if __name__ == "__main__":
    start_metrics_server(9090)
    run_scraper()

Or if you're using Flask/FastAPI:

# FastAPI example
from fastapi import FastAPI
from prometheus_client import generate_latest
from starlette.responses import Response

app = FastAPI()

@app.get("/metrics")async def metrics():
    return Response(
        generate_latest(),
        media_type="text/plain; charset=utf-8"
    )

Step 4: Prometheus Configuration

# prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "captcha-solver"
    static_configs:
      - targets: ["scraper-host:9090"]
    scrape_interval: 10s

  # If running multiple scrapers
  - job_name: "captcha-solver-fleet"
    dns_sd_configs:
      - names: ["scraper.service.consul"]
        type: "SRV"

Step 5: Grafana Dashboard

Here are the key panels for your CAPTCHA monitoring dashboard:

Panel 1: Success Rate (should be > 90%)

# Success rate over last 5 minutes
sum(rate(
    captcha_solves_total{status="success"}[5m]
)) by (captcha_type) 
/ 
sum(rate(
    captcha_solves_total[5m]
)) by (captcha_type)

Panel 2: Solve Latency (p95)

histogram_quantile(0.95, 
    sum(rate(
        captcha_solve_duration_seconds_bucket[5m]
    )) by (le, captcha_type)
)

Panel 3: Error Breakdown

sum(rate(
    captcha_errors_total[5m]
)) by (captcha_type, error_type)

Panel 4: Cost Tracking

# Cost per hour by type
sum(rate(
    captcha_cost_dollars_total[1h]
)) by (captcha_type) * 3600

Panel 5: Concurrent Solves (capacity gauge)

captcha_concurrent_solves

Step 6: Alert Rules

Set up alerts before things break:

# alert_rules.yml
groups:
  - name: captcha_alerts
    rules:
      # Success rate drops below 80%
      - alert: CaptchaSolveRateLow
        expr: |
          sum(rate(captcha_solves_total{status="success"}[10m]))
          /
          sum(rate(captcha_solves_total[10m]))
          < 0.8
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "CAPTCHA solve rate below 80%"
          description: >
            Current rate: {{ $value | humanizePercentage }}

      # P95 latency above 30 seconds
      - alert: CaptchaSolveLatencyHigh
        expr: |
          histogram_quantile(0.95,
            sum(rate(
              captcha_solve_duration_seconds_bucket[5m]
            )) by (le)
          ) > 30
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "CAPTCHA solve p95 latency above 30s"

      # No solves happening at all
      - alert: CaptchaSolvesStopped
        expr: |
          sum(rate(captcha_solves_total[5m])) == 0
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "No CAPTCHA solves in the last 10 minutes"

      # Cost spike detection
      - alert: CaptchaCostSpike
        expr: |
          sum(rate(captcha_cost_dollars_total[1h])) * 3600
          > 5.0
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "CAPTCHA cost exceeding $5/hour"

Step 7: Structured Logging

Metrics tell you what happened. Logs tell you why:

# logging_config.py
import structlog
import logging

def setup_logging():
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(
                fmt="iso"
            ),
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

logger = structlog.get_logger()

# In your solver
def solve(self, captcha_type, sitekey, url):
    log = logger.bind(
        captcha_type=captcha_type,
        target_url=url
    )

    log.info("captcha.solve.started")

    try:
        token = self._do_solve(captcha_type, sitekey, url)
        log.info(
            "captcha.solve.success",
            duration=duration,
            token_length=len(token)
        )
        return token
    except Exception as e:
        log.error(
            "captcha.solve.failed",
            error=str(e),
            error_type=type(e).__name__
        )
        raise

Output:

{"event": "captcha.solve.started", "captcha_type": "recaptcha_v2", "target_url": "https://example.com", "timestamp": "2025-03-23T10:15:00Z", "level": "info"}
{"event": "captcha.solve.success", "captcha_type": "recaptcha_v2", "duration": 4.2, "token_length": 580, "timestamp": "2025-03-23T10:15:04Z", "level": "info"}

Step 8: Health Check Endpoint

Add a health check that verifies your solver is working:

@app.get("/health")async def health():
    # Try a test solve
    try:
        test_sitekey = "6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI"
        start = time.time()
        solver.solve(
            "recaptcha_v2", 
            test_sitekey, 
            "https://example.com"
        )
        latency = time.time() - start

        return {
            "status": "healthy",
            "solve_latency": f"{latency:.1f}s",
            "stats": solver.stats
        }
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "error": str(e)
            }
        )

Quick Start: Minimal Setup

Don't want the full Prometheus/Grafana stack? Start with just logging:

# Minimal monitoring — just print stats every 60s
import asyncio

async def stats_reporter(solver, interval=60):
    while True:
        await asyncio.sleep(interval)
        stats = solver.stats
        rate = (
            stats["solved"] / 
            (stats["solved"] + stats["failed"]) * 100
            if (stats["solved"] + stats["failed"]) > 0 
            else 0
        )
        print(
            f"[CAPTCHA Stats] "
            f"Solved: {stats['solved']} | "
            f"Failed: {stats['failed']} | "
            f"Rate: {rate:.1f}% | "
            f"Avg latency: {stats.get('avg_latency', 0):.1f}s"
        )

Key Takeaways

Instrument from day one — adding metrics later is painful
Track success rate as your primary metric — everything else is secondary
Alert on rate drops, not individual failures — transient errors are normal
Separate concerns — metrics for dashboards, structured logs for debugging
Cost tracking prevents surprises — alert before the bill does

For the CAPTCHA-solving client used in these examples, check out passxapi-python. It provides clean interfaces that are easy to wrap with your own metrics layer.

What monitoring stack do you use for your scrapers? I'd love to hear what metrics you track.

Building a High-Throughput Async Scraper with aiohttp and Automatic CAPTCHA Handling

Alex Chen — Mon, 23 Mar 2026 00:25:34 +0000

You need to scrape 10,000 product pages. With requests\, that's ~14 hours (5s per page). With aiohttp\ and proper concurrency control, it's under 30 minutes.

But here's the catch: at scale, you will hit CAPTCHAs. Let's build an async scraper from scratch that handles them transparently.

Why aiohttp Over requests/httpx?

Quick comparison for I/O-bound scraping:

Library	1000 pages	Async	Memory
requests	~83 min	No	Low
httpx (async)	~8 min	Yes	Medium
aiohttp	~5 min	Yes	Low

aiohttp is purpose-built for async HTTP. It reuses connections aggressively and has lower overhead per request than httpx.

Project Structure

async_scraper/
├── scraper/
│   ├── __init__.py
│   ├── client.py        # aiohttp session management
│   ├── captcha.py       # CAPTCHA detection + solving
│   ├── pipeline.py      # Data processing pipeline
│   └── rate_limiter.py  # Concurrency + rate control
├── config.py
├── main.py
└── requirements.txt

Step 1: The Rate Limiter

Don't blast servers with 1000 concurrent requests. Use a token bucket:

# scraper/rate_limiter.py
import asyncio
import time

class TokenBucket:
    """Rate limiter using token bucket algorithm."""

    def __init__(
        self, 
        rate: float,        # tokens per second
        max_tokens: int = 10
    ):
        self.rate = rate
        self.max_tokens = max_tokens
        self._tokens = max_tokens
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(
                self.max_tokens,
                self._tokens + elapsed * self.rate
            )
            self._last_refill = now

            if self._tokens < 1:
                wait_time = (1 - self._tokens) / self.rate
                await asyncio.sleep(wait_time)
                self._tokens = 0
            else:
                self._tokens -= 1


class ConcurrencyLimiter:
    """Combines semaphore + rate limiting."""

    def __init__(
        self, 
        max_concurrent: int = 20,
        requests_per_second: float = 10.0
    ):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.bucket = TokenBucket(
            rate=requests_per_second
        )

    async def __aenter__(self):
        await self.semaphore.acquire()
        await self.bucket.acquire()
        return self

    async def __aexit__(self, *args):
        self.semaphore.release()

Step 2: CAPTCHA Detection and Solving

Build a detector that works across CAPTCHA types:

# scraper/captcha.py
import re
from dataclasses import dataclass
from enum import Enum

class CaptchaType(Enum):
    RECAPTCHA_V2 = "recaptcha_v2"
    RECAPTCHA_V3 = "recaptcha_v3"
    HCAPTCHA = "hcaptcha"
    TURNSTILE = "turnstile"
    NONE = "none"

@dataclassclass CaptchaInfo:
    type: CaptchaType
    sitekey: str
    page_url: str
    extra: dict = None

CAPTCHA_PATTERNS = [
    {
        "type": CaptchaType.RECAPTCHA_V2,
        "selector": r'class="g-recaptcha"[^>]*data-sitekey="([^"]+)"',
    },
    {
        "type": CaptchaType.RECAPTCHA_V3,
        "selector": r'grecaptcha\.execute\([\'"]([^\'"]+)',
    },
    {
        "type": CaptchaType.HCAPTCHA,
        "selector": r'class="h-captcha"[^>]*data-sitekey="([^"]+)"',
    },
    {
        "type": CaptchaType.TURNSTILE,
        "selector": r'class="cf-turnstile"[^>]*data-sitekey="([^"]+)"',
    },
]

def detect_captcha(html: str, url: str) -> CaptchaInfo | None:
    for pattern in CAPTCHA_PATTERNS:
        match = re.search(pattern["selector"], html)
        if match:
            return CaptchaInfo(
                type=pattern["type"],
                sitekey=match.group(1),
                page_url=url
            )
    return None

Now the async solver:

# scraper/captcha.py (continued)
import aiohttp
import asyncio

class AsyncCaptchaSolver:
    def __init__(
        self, 
        api_base: str = "https://www.passxapi.com",
        max_concurrent_solves: int = 5
    ):
        self.api_base = api_base
        self.semaphore = asyncio.Semaphore(
            max_concurrent_solves
        )
        self.stats = {"solved": 0, "failed": 0}

    async def solve(
        self, 
        info: CaptchaInfo,
        session: aiohttp.ClientSession
    ) -> str | None:
        async with self.semaphore:
            try:
                # Submit task
                async with session.post(
                    f"{self.api_base}/api/v1/task",
                    json={
                        "type": info.type.value,
                        "sitekey": info.sitekey,
                        "pageurl": info.page_url,
                        **(info.extra or {})
                    }
                ) as resp:
                    data = await resp.json()
                    task_id = data["task_id"]

                # Poll for result
                for _ in range(60):
                    async with session.get(
                        f"{self.api_base}/api/v1/task/{task_id}"
                    ) as resp:
                        result = await resp.json()
                        if result["status"] == "completed":
                            self.stats["solved"] += 1
                            return result["token"]
                        if result["status"] == "failed":
                            raise Exception(result.get("error"))
                    await asyncio.sleep(2)

                raise TimeoutError("Solve timed out")

            except Exception as e:
                self.stats["failed"] += 1
                print(f"CAPTCHA solve failed: {e}")
                return None

Step 3: The Main Scraper Client

# scraper/client.py
import aiohttp
import asyncio
from .captcha import detect_captcha, AsyncCaptchaSolver
from .rate_limiter import ConcurrencyLimiter

class AsyncScraper:
    def __init__(
        self,
        max_concurrent: int = 20,
        requests_per_second: float = 10.0,
        max_retries: int = 3
    ):
        self.limiter = ConcurrencyLimiter(
            max_concurrent, requests_per_second
        )
        self.solver = AsyncCaptchaSolver()
        self.max_retries = max_retries
        self.session = None
        self._results = []

    async def __aenter__(self):
        connector = aiohttp.TCPConnector(
            limit=100,           # Connection pool size
            ttl_dns_cache=300,   # DNS cache 5 min
            enable_cleanup_closed=True
        )
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(total=30),
            headers={
                "User-Agent": (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36"
                )
            }
        )
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    async def fetch(self, url: str) -> dict:
        """Fetch a URL with CAPTCHA handling."""
        for attempt in range(self.max_retries):
            async with self.limiter:
                try:
                    async with self.session.get(url) as resp:
                        html = await resp.text()

                        # Check for CAPTCHA
                        captcha = detect_captcha(html, url)
                        if captcha:
                            token = await self.solver.solve(
                                captcha, self.session
                            )
                            if token:
                                # Resubmit with token
                                html = await self._submit_with_token(
                                    url, token, captcha
                                )

                        return {
                            "url": url,
                            "status": resp.status,
                            "html": html,
                            "captcha_solved": captcha is not None
                        }

                except (
                    aiohttp.ClientError,
                    asyncio.TimeoutError
                ) as e:
                    if attempt == self.max_retries - 1:
                        return {
                            "url": url, "error": str(e)
                        }
                    # Exponential backoff
                    await asyncio.sleep(2 ** attempt)

    async def _submit_with_token(
        self, url, token, captcha_info
    ):
        """POST the solved token back to the page."""
        field_names = {
            "recaptcha_v2": "g-recaptcha-response",
            "recaptcha_v3": "g-recaptcha-response",
            "hcaptcha": "h-captcha-response",
            "turnstile": "cf-turnstile-response",
        }
        field = field_names.get(
            captcha_info.type.value, 
            "captcha_response"
        )

        async with self.session.post(
            url, data={field: token}
        ) as resp:
            return await resp.text()

    async def scrape_many(
        self, urls: list[str]
    ) -> list[dict]:
        """Scrape multiple URLs concurrently."""
        tasks = [self.fetch(url) for url in urls]
        results = await asyncio.gather(
            *tasks, return_exceptions=True
        )

        # Convert exceptions to error dicts
        processed = []
        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                processed.append({
                    "url": url, "error": str(result)
                })
            else:
                processed.append(result)

        return processed

Step 4: Data Processing Pipeline

Process results as they come in, don't wait for all pages:

# scraper/pipeline.py
import asyncio
import json
from pathlib import Path

class DataPipeline:
    def __init__(self, output_dir: str = "output"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.queue = asyncio.Queue()
        self.processed = 0

    async def process(self, result: dict):
        """Extract and save data from a scraped page."""
        if "error" in result:
            print(f"  ✗ {result['url']}: {result['error']}")
            return

        # Your extraction logic here
        data = self.extract(result["html"], result["url"])

        if data:
            await self.queue.put(data)
            self.processed += 1

    def extract(self, html: str, url: str) -> dict | None:
        """Override this with your parsing logic."""
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        title = soup.find("h1")
        price = soup.find(class_="price")

        if not title:
            return None

        return {
            "url": url,
            "title": title.text.strip(),
            "price": price.text.strip() if price else None,
        }

    async def writer(self):
        """Background task that writes results to disk."""
        output_file = self.output_dir / "results.jsonl"

        with open(output_file, "a") as f:
            while True:
                try:
                    data = await asyncio.wait_for(
                        self.queue.get(), timeout=5.0
                    )
                    f.write(json.dumps(data) + "\n")
                    f.flush()
                except asyncio.TimeoutError:
                    continue
                except asyncio.CancelledError:
                    # Drain remaining items
                    while not self.queue.empty():
                        data = self.queue.get_nowait()
                        f.write(json.dumps(data) + "\n")
                    break

Step 5: Putting It All Together

# main.py
import asyncio
import time
from scraper.client import AsyncScraper
from scraper.pipeline import DataPipeline

async def main():
    urls = [
        f"https://example-shop.com/product/{i}"
        for i in range(1, 10001)
    ]

    pipeline = DataPipeline(output_dir="output")
    writer_task = asyncio.create_task(pipeline.writer())

    print(f"Scraping {len(urls)} URLs...")
    start = time.monotonic()

    async with AsyncScraper(
        max_concurrent=20,
        requests_per_second=10
    ) as scraper:
        # Process in batches of 100
        batch_size = 100
        for i in range(0, len(urls), batch_size):
            batch = urls[i:i + batch_size]
            results = await scraper.scrape_many(batch)

            for result in results:
                await pipeline.process(result)

            elapsed = time.monotonic() - start
            rate = (i + len(batch)) / elapsed
            print(
                f"  Batch {i // batch_size + 1}: "
                f"{rate:.0f} pages/min | "
                f"CAPTCHAs solved: "
                f"{scraper.solver.stats['solved']}"
            )

    # Stop the writer
    writer_task.cancel()
    try:
        await writer_task
    except asyncio.CancelledError:
        pass

    elapsed = time.monotonic() - start
    print(f"\nDone! {len(urls)} pages in {elapsed:.0f}s")
    print(f"Extracted: {pipeline.processed} items")
    print(f"CAPTCHA stats: {scraper.solver.stats}")


if __name__ == "__main__":
    asyncio.run(main())

Performance Tuning Tips

1. Connection Pool Size Matters

# Too low: connections wait in queue
connector = aiohttp.TCPConnector(limit=10)   # Slow

# Sweet spot: matches your concurrency
connector = aiohttp.TCPConnector(limit=100)  # Good

# Too high: target server may block you
connector = aiohttp.TCPConnector(limit=1000) # Risky

2. DNS Caching Saves Milliseconds Per Request

# Without cache: DNS lookup every request (~50ms each)
# With cache: DNS lookup once per domain
connector = aiohttp.TCPConnector(
    ttl_dns_cache=300,   # Cache for 5 minutes
    use_dns_cache=True   # Enabled by default
)

3. Separate CAPTCHA Solve Concurrency

Don't let CAPTCHA solves eat all your concurrent slots:

# Main scraping: 20 concurrent
scraper_sem = asyncio.Semaphore(20)

# CAPTCHA solving: 5 concurrent (separate pool)
solver_sem = asyncio.Semaphore(5)

4. Monitor and Adjust in Real-Time

async def adaptive_rate(scraper, initial_rate=10):
    """Adjust rate based on error ratio."""
    rate = initial_rate
    while True:
        await asyncio.sleep(60)
        stats = scraper.solver.stats
        total = stats["solved"] + stats["failed"]
        if total > 0:
            error_rate = stats["failed"] / total
            if error_rate > 0.3:
                rate *= 0.7  # Back off
            elif error_rate < 0.05:
                rate *= 1.2  # Speed up
        scraper.limiter.bucket.rate = rate

Expected Performance

With 20 concurrent connections and 10 req/s:

Pages	No CAPTCHAs	10% CAPTCHA rate	50% CAPTCHA rate
100	~10s	~15s	~45s
1,000	~100s	~150s	~450s
10,000	~17min	~25min	~75min

The CAPTCHA overhead scales with hit rate, not total pages — because solves happen concurrently with other fetches.

Wrapping Up

The key patterns for high-throughput async scraping:

Token bucket for rate limiting (not just semaphore)
Separate concurrency pools for fetching vs CAPTCHA solving
Streaming pipeline — process data as it arrives
Adaptive rate control — respond to errors dynamically

For the CAPTCHA-solving integration used in these examples, check out passxapi-python — it provides both sync and async clients that plug into aiohttp workflows.

What throughput are you getting with your scrapers? Drop your numbers in the comments.

Testing Your Scraper: How to Write pytest Tests for Code That Hits CAPTCHAs

Alex Chen — Mon, 23 Mar 2026 00:18:48 +0000

You built a scraper that handles CAPTCHAs. It works. But how do you test it without burning money on solves every time you run your test suite?

This is a real problem. You need tests that are fast, cheap, and still catch real bugs. Let's build a testing strategy that covers all the layers — from unit tests with mocks to integration tests with real solves.

The Testing Pyramid for Scrapers

        /\
       /  \    E2E (real sites + real solves)
      /----\   
     /      \  Integration (real API, test sitekeys)
    /--------\
   /          \ Unit (mocked everything)
  /____________\

Most of your tests should be unit tests. A few integration tests. Rare E2E tests (maybe in CI nightly).

Layer 1: Unit Tests with Mocked CAPTCHA Responses

Test your scraper logic without any network calls:

# tests/test_scraper_unit.py
import pytest
from unittest.mock import AsyncMock, patch

from scraper import PageScraper
from captcha import CaptchaToken

@pytest.fixturedef mock_solver():
    """Mock CAPTCHA solver that returns instantly."""
    solver = AsyncMock()
    solver.solve.return_value = CaptchaToken(
        value="mock-token-abc123",
        ttl=120
    )
    return solver

@pytest.fixturedef scraper(mock_solver):
    return PageScraper(captcha_solver=mock_solver)


class TestCaptchaDetection:
    """Test that we correctly identify CAPTCHAs in HTML."""

    def test_detects_recaptcha_v2(self, scraper):
        html = """
        <div class="g-recaptcha" 
             data-sitekey="6Le-wvkSAAAAAPBM...">
        </div>
        """
        result = scraper.detect_captcha(html)
        assert result["type"] == "recaptcha_v2"
        assert result["sitekey"] == "6Le-wvkSAAAAAPBM..."

    def test_detects_hcaptcha(self, scraper):
        html = """
        <div class="h-captcha" 
             data-sitekey="abc-123-def">
        </div>
        """
        result = scraper.detect_captcha(html)
        assert result["type"] == "hcaptcha"

    def test_detects_turnstile(self, scraper):
        html = """
        <div class="cf-turnstile" 
             data-sitekey="0x4AAAAAAA...">
        </div>
        """
        result = scraper.detect_captcha(html)
        assert result["type"] == "turnstile"

    def test_no_captcha_returns_none(self, scraper):
        html = "<html><body><form>No captcha here</form></body></html>"
        assert scraper.detect_captcha(html) is None


class TestTokenInjection:
    """Test that tokens get injected correctly."""

    @pytest.mark.asyncio    async def test_injects_token_into_form(self, scraper, mock_solver):
        # Mock the page response
        with patch.object(
            scraper, "fetch_page", 
            return_value=SAMPLE_HTML_WITH_CAPTCHA
        ):
            result = await scraper.scrape("https://example.com")

            # Verify solver was called with correct params
            mock_solver.solve.assert_called_once_with(
                captcha_type="recaptcha_v2",
                sitekey="6Le-wvkSAAAAAPBM...",
                url="https://example.com"
            )

    @pytest.mark.asyncio    async def test_skips_solve_when_no_captcha(
        self, scraper, mock_solver
    ):
        with patch.object(
            scraper, "fetch_page", 
            return_value="<html>No captcha</html>"
        ):
            await scraper.scrape("https://example.com")
            mock_solver.solve.assert_not_called()

Layer 2: Testing Error Recovery

Your retry logic is critical. Test every failure mode:

# tests/test_error_handling.py
import pytest
from unittest.mock import AsyncMock
from captcha import (
    CaptchaSolver, 
    TimeoutError, 
    InvalidTypeError,
    TokenExpiredError
)

class TestRetryBehavior:

    @pytest.mark.asyncio    async def test_retries_on_timeout(self):
        solver = CaptchaSolver()
        solver._do_solve = AsyncMock(
            side_effect=[
                TimeoutError("Timed out"),
                TimeoutError("Timed out"),
                "valid-token-123"  # Succeeds on 3rd try
            ]
        )

        token = await solver.solve(
            "recaptcha_v2", "sitekey", "https://example.com"
        )
        assert token.value == "valid-token-123"
        assert solver._do_solve.call_count == 3

    @pytest.mark.asyncio    async def test_fails_after_max_retries(self):
        solver = CaptchaSolver(max_retries=3)
        solver._do_solve = AsyncMock(
            side_effect=TimeoutError("Always fails")
        )

        with pytest.raises(TimeoutError):
            await solver.solve(
                "recaptcha_v2", "sitekey", 
                "https://example.com"
            )
        assert solver._do_solve.call_count == 3

    @pytest.mark.asyncio    async def test_no_retry_on_invalid_api_key(self):
        """Fatal errors should not be retried."""
        solver = CaptchaSolver()
        solver._do_solve = AsyncMock(
            side_effect=AuthenticationError("Invalid API key")
        )

        with pytest.raises(AuthenticationError):
            await solver.solve(
                "recaptcha_v2", "sitekey", 
                "https://example.com"
            )
        # Should fail immediately, no retries
        assert solver._do_solve.call_count == 1


class TestCircuitBreaker:

    @pytest.mark.asyncio    async def test_opens_after_threshold(self):
        solver = CaptchaSolver(
            circuit_failure_threshold=3
        )
        solver._do_solve = AsyncMock(
            side_effect=TimeoutError("Down")
        )

        # Fail 3 times to trip the circuit
        for _ in range(3):
            with pytest.raises(TimeoutError):
                await solver.solve(
                    "recaptcha_v2", "sk", "https://ex.com"
                )

        # 4th call should fail immediately (circuit open)
        with pytest.raises(CircuitOpenError):
            await solver.solve(
                "recaptcha_v2", "sk", "https://ex.com"
            )

Layer 3: Testing Token Expiry Logic

Tokens expire. Your tests should verify you handle that:

# tests/test_token_freshness.py
import pytest
import time
from unittest.mock import patch
from captcha import CaptchaToken

class TestTokenExpiry:

    def test_fresh_token_is_valid(self):
        token = CaptchaToken(value="abc", ttl=120)
        assert token.is_valid is True
        assert token.remaining_seconds > 119

    def test_expired_token_is_invalid(self):
        token = CaptchaToken(value="abc", ttl=120)
        # Fast-forward time
        with patch("time.time", return_value=time.time() + 121):
            assert token.is_valid is False
            assert token.remaining_seconds == 0

    def test_almost_expired_token(self):
        """Token with < 10s left should be treated 
        as expired for safety."""
        token = CaptchaToken(value="abc", ttl=120)
        with patch(
            "time.time", 
            return_value=time.time() + 115
        ):
            assert token.remaining_seconds < 10
            # Your code should re-solve, not submit

Layer 4: Integration Tests with Test Sitekeys

Google and hCaptcha provide test sitekeys that always pass. Use them for integration tests:

# tests/test_integration.py
import pytest
import os

# Skip if no API key (in CI, set as secret)
pytestmark = pytest.mark.skipif(
    not os.getenv("PASSXAPI_KEY"),
    reason="No API key for integration tests"
)

# Official test sitekeys (always solvable)
TEST_SITEKEYS = {
    "recaptcha_v2": "6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI",
    "hcaptcha": "10000000-ffff-ffff-ffff-000000000001",
}

class TestRealSolves:
    """Integration tests that hit the real API.
    Run sparingly — these cost money."""

    @pytest.mark.asyncio    @pytest.mark.slow    async def test_solve_recaptcha_v2_test_key(self):
        import httpx

        async with httpx.AsyncClient(
            base_url="https://www.passxapi.com"
        ) as client:
            resp = await client.post("/api/v1/task", json={
                "type": "recaptcha_v2",
                "sitekey": TEST_SITEKEYS["recaptcha_v2"],
                "pageurl": "https://example.com"
            })
            task = resp.json()
            assert "task_id" in task

            # Poll for result
            import asyncio
            for _ in range(30):
                result = await client.get(
                    f"/api/v1/task/{task['task_id']}"
                )
                data = result.json()
                if data["status"] == "completed":
                    assert len(data["token"]) > 50
                    return
                await asyncio.sleep(2)

            pytest.fail("Solve timed out")

    @pytest.mark.asyncio    @pytest.mark.slow    async def test_solve_hcaptcha_test_key(self):
        # Similar pattern with hCaptcha test key
        ...

Layer 5: Fixture Factory Pattern

Create reusable fixtures for common CAPTCHA scenarios:

# tests/conftest.py
import pytest
from unittest.mock import AsyncMock

@pytest.fixturedef captcha_html_factory():
    """Generate HTML pages with different CAPTCHAs."""
    def _make(
        captcha_type: str = "recaptcha_v2",
        sitekey: str = "test-key-123"
    ) -> str:
        templates = {
            "recaptcha_v2": f"""
                <div class="g-recaptcha" 
                     data-sitekey="{sitekey}"></div>
            """,
            "hcaptcha": f"""
                <div class="h-captcha" 
                     data-sitekey="{sitekey}"></div>
            """,
            "turnstile": f"""
                <div class="cf-turnstile" 
                     data-sitekey="{sitekey}"></div>
            """,
            "none": "<form><input type='submit'></form>"
        }
        body = templates.get(captcha_type, templates["none"])
        return f"<html><body>{body}</body></html>"

    return _make


@pytest.fixturedef slow_solver():
    """Solver that simulates network delay."""
    import asyncio

    async def _slow_solve(*args, **kwargs):
        await asyncio.sleep(0.1)  # Simulate API latency
        return "delayed-token-xyz"

    solver = AsyncMock(side_effect=_slow_solve)
    return solver


@pytest.fixturedef flaky_solver():
    """Solver that fails 50% of the time."""
    call_count = 0

    async def _flaky(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count % 2 == 0:
            raise TimeoutError("Randomly failed")
        return f"token-attempt-{call_count}"

    return AsyncMock(side_effect=_flaky)

Running Your Test Suite

Set up pytest.ini to separate fast and slow tests:

# pytest.ini
[pytest]
markers =
    slow: integration tests that hit real APIs

# Default: skip slow tests
addopts = -m "not slow" --tb=short -q

# Fast unit tests (daily, in pre-commit)
pytest

# Full suite with integration tests (nightly CI)
PASSXAPI_KEY=your_key pytest -m "" --tb=long

# Just the CAPTCHA detection tests
pytest tests/test_scraper_unit.py::TestCaptchaDetection -v

Key Takeaways

Unit test your detection logic — parse HTML, find sitekeys, classify types
Mock the solver for speed — don't hit APIs in unit tests
Test every failure mode — timeout, wrong type, expired tokens, circuit breaker
Use test sitekeys for integration tests — they're free and always solvable
Separate fast/slow tests with pytest markers
Fixture factories make it easy to generate test scenarios

For the CAPTCHA-solving client used in these examples, check out passxapi-python — it provides a clean async interface that's easy to mock in tests.

How do you test your scrapers? Share your approach in the comments.