Requirements:

pip install requests beautifulsoup4

import time

from typing import List, Dict

import requests

from bs4 import BeautifulSoup

HEADERS = {"User-Agent": "MasteringClaudeCode/1.0 (+https://example.org)"}

def fetch(url: str, timeout: float = 10.0) -> str:

"""Fetch a URL synchronously and return text content."""

resp = requests.get(url, headers=HEADERS, timeout=timeout)

resp.raise_for_status()

return resp.text

def parse(html: str) -> Dict[str, str]:

"""Extract simple fields: <title> and first <h1> if present."""

soup = BeautifulSoup(html, "html.parser")

title = (soup.title.string or "").strip() if soup.title else ""

h1 = (soup.find("h1").get_text(strip=True)) if soup.find("h1") else ""

return {"title": title, "h1": h1}

def scrape(urls: List[str]) -> List[Dict[str, str]]:

results = []

for url in urls:

    html = fetch(url)

    data = parse(html)

    data["url"] = url

    results.append(data)

    time.sleep(0.2) # polite delay

return results

if name == "main":

URLS = [

    "https://example.com",

    "https://httpbin.org/html",

    "https://www.iana.org/domains/reserved",

]

start = time.time()

items = scrape(URLS)

elapsed = time.time() - start

for item in items:

    print(f"{item['url']}\n title: {item['title']}\n h1: {item['h1']}\n")

print(f"Baseline elapsed: {elapsed:.3f}s")

This baseline is simple and correct, but it blocks on each network request. As the URL list grows, total time scales linearly with latency, and any transient error stops progress.

Optimized: Concurrent, Pooled, and Robust Scraper

```python

# optimized_scraper.py

# Requirements:

# pip install aiohttp aiodns beautifulsoup4 cchardet orjson

# (aiodns optional; speeds up DNS on some platforms)
import asyncio

from typing import List, Dict, Optional, Tuple

from contextlib import asynccontextmanager

import random

import time

import aiohttp

from bs4 import BeautifulSoup

HEADERS = {"User-Agent": "MasteringClaudeCode/1.0 (+https://example.org)"}

Bounded concurrency avoids overwhelming remote servers and your machine.

MAX_CONCURRENCY = 10

REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=12, connect=5, sock_connect=5, sock_read=10)

@asynccontextmanager

async def make_session() -> aiohttp.ClientSession:

connector = aiohttp.TCPConnector(limit=MAX_CONCURRENCY, ttl_dns_cache=300)

async with aiohttp.ClientSession(headers=HEADERS, timeout=REQUEST_TIMEOUT, connector=connector) as session:

yield session

async def fetch(session: aiohttp.ClientSession, url: str, attempt: int = 1, max_attempts: int = 3) -> str:

"""

Fetch a URL with exponential backoff on transient errors.

Raises the last exception if all attempts fail.

"""

try:

async with session.get(url) as resp:

Fail fast on non-2xx responses

if resp.status >= 400:

raise aiohttp.ClientResponseError(

request_info=resp.request_info,

history=resp.history,

status=resp.status,

message=f"HTTP {resp.status}",

headers=resp.headers,

)

Decode efficiently; rely on aiohttp's chardet if needed

return await resp.text()

except (aiohttp.ClientError, asyncio.TimeoutError) as e:

if attempt >= max_attempts:

raise

Exponential backoff with jitter

backoff = (2 ** (attempt - 1)) + random.uniform(0, 0.25)

await asyncio.sleep(backoff)

return await fetch(session, url, attempt + 1, max_attempts)

def parse_fast(html: str) -> Dict[str, str]:

"""

Parse using 'lxml' if available via BeautifulSoup for speed,

falling back to 'html.parser' otherwise.

"""

try:

soup = BeautifulSoup(html, "lxml")

except Exception:

soup = BeautifulSoup(html, "html.parser")

title = (soup.title.string or "").strip() if soup.title else ""

h1_tag = soup.find("h1")

h1 = h1_tag.get_text(strip=True) if h1_tag else ""

return {"title": title, "h1": h1}

async def scrape_one(semaphore: asyncio.Semaphore, session: aiohttp.ClientSession, url: str) -> Tuple[str, Optional[Dict[str, str]], Optional[str]]:

async with semaphore:

try:

html = await fetch(session, url)

data = parse_fast(html)

data["url"] = url

return url, data, None

except Exception as e:

return url, None, f"{type(e).name}: {e}"

async def scrape(urls: List[str]) -> List[Dict[str, str]]:

semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

async with make_session() as session:

tasks = [scrape_one(semaphore, session, url) for url in urls]

results = []

for coro in asyncio.as_completed(tasks):

url, data, err = await coro

if err:

results.append({"url": url, "error": err, "title": "", "h1": ""})

else:

results.append(data)

return results

if name == "main":

URLS = [

"https://example.com",

"https://httpbin.org/html",

"https://www.iana.org/domains/reserved",

Add more URLs here to see concurrency shine

]

started = time.time()

items = asyncio.run(scrape(URLS))

elapsed = time.time() - started

for item in items:

if item.get("error"):

print(f"{item['url']}\n error: {item['error']}\n")

else:

print(f"{item['url']}\n title: {item['title']}\n h1: {item['h1']}\n")

print(f"Optimized elapsed: {elapsed:.3f}s with concurrency={MAX_CONCURRENCY}")

Key improvements in the optimized version:

• Bounded concurrency with a semaphore and a connector limit ensures politeness and protects resources.
• Connection pooling via TCPConnector reduces handshake overhead across many requests.
• Timeouts and backoff prevent stuck requests and recover from transient failures without crashing the entire run.
• Faster parsing prefers lxml automatically while remaining compatible with the standard parser.
• Structured results capture successes and errors uniformly, keeping downstream processing simple.

Clarification Table

Optimization	What Changed	Why It Matters	Impact on Behavior
Concurrency	Single-thread loop → asyncio with aiohttp and semaphore	Hides network latency by overlapping I/O	Dramatically lowers wall-clock time for many URLs
Connection Reuse	One TCP connection per request → pooled TCPConnector	Reduces repeated handshakes and DNS lookups	Cuts per-request overhead
Timeouts & Backoff	No timeouts/retries → explicit timeouts + exponential backoff	Fails fast and recovers from transient errors	Improves robustness and throughput stability
Parser Choice	Built-in parser only → prefer lxml with fallback	Faster HTML parsing in CPython	Reduces CPU time per page
Error Handling	Exceptions bubble → structured error rows	Keeps the pipeline running on partial failures	Better operability at scale

You started with a correct synchronous scraper and, by reasoning about where time is truly spent, evolved it into a concurrent, pooled, and fault-tolerant pipeline that remains easy to read and modify. Claude Code’s value in this workflow is the ability to explain why each change matters, predict its impact, and keep the design clean as performance improves. In the next section, you will generalize these techniques into a reusable optimization checklist for data-collection services: measuring bottlenecks, choosing the right concurrency model, and validating that improved speed doesn’t compromise correctness or maintainability.