Tech Tutorial - February 17 2026 233008

Welcome back, Codeyaan explorers! Today we’ll dive deep into asynchronous web scraping with Python 3.12, the httpx library, and BeautifulSoup. By the end of this tutorial you’ll have a production‑ready scraper that can fetch hundreds of pages per minute, respect rate limits, and gracefully recover from hiccups—all while keeping your code clean and maintainable.

Why Asynchronous Scraping Matters

Traditional synchronous scrapers block on each HTTP request, which means the CPU sits idle while waiting for the network. In contrast, asynchronous I/O lets a single thread juggle dozens of connections at once, dramatically boosting throughput without the overhead of multi‑processing or heavy thread pools.

Real‑world scenarios—price monitoring, SEO audits, or large‑scale data collection—often demand thousands of requests per run. Using asyncio + httpx gives you the speed of a multi‑core solution while staying memory‑light.

Setting Up the Environment

First, create a fresh virtual environment. This isolates dependencies and ensures reproducibility across machines.

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install --upgrade pip
pip install httpx[http2] beautifulsoup4 lxml tqdm

We install httpx[http2] to enable HTTP/2 where the target site supports it, beautifulsoup4 for parsing, lxml as the fast parser backend, and tqdm for a lightweight progress bar.

Core Concepts You’ll Use

1. Async Context Managers

Both httpx.AsyncClient and file handles can be used as async context managers, guaranteeing proper cleanup even when exceptions occur.

2. Semaphore for Rate Limiting

A semaphore caps the number of concurrent requests, preventing you from overwhelming the target server and getting blocked.

3. Exponential Back‑off

When a request fails with a 429 or 5xx status, we retry after a delay that grows exponentially. This pattern is polite and improves overall success rates.

Example 1: A Minimal Async Scraper

Let’s start with the simplest possible asynchronous scraper: fetch a list of URLs, parse the page title, and store results in a dictionary.

import asyncio
import httpx
from bs4 import BeautifulSoup

async def fetch_title(client: httpx.AsyncClient, url: str) -> str:
    resp = await client.get(url, timeout=10.0)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "lxml")
    return soup.title.string.strip() if soup.title else "No title"

async def main(urls: list[str]) -> dict[str, str]:
    async with httpx.AsyncClient(http2=True) as client:
        tasks = [fetch_title(client, u) for u in urls]
        titles = await asyncio.gather(*tasks, return_exceptions=True)

    return {url: (title if not isinstance(title, Exception) else str(title))
            for url, title in zip(urls, titles)}

if __name__ == "__main__":
    sample_urls = [
        "https://www.python.org",
        "https://realpython.com",
        "https://news.ycombinator.com",
    ]
    result = asyncio.run(main(sample_urls))
    for u, t in result.items():
        print(f"{u} → {t}")

This script demonstrates the essential async flow: create a client, launch a bunch of coroutines, and await them all together. The return_exceptions=True flag ensures that a single failing request won’t abort the whole batch.

Example 2: Scaling Up with Rate Limiting & Retries

For production use we need to add three layers of robustness:

1. Concurrency control via an asyncio.Semaphore.
2. Retry logic with exponential back‑off.
3. Progress reporting using tqdm.

import asyncio
import random
import time
from typing import List

import httpx
from bs4 import BeautifulSoup
from tqdm.asyncio import tqdm_asyncio

MAX_CONCURRENCY = 20          # How many requests at once
MAX_RETRIES = 5               # Maximum retry attempts per URL
BASE_DELAY = 0.5              # Initial back‑off delay (seconds)

sem = asyncio.Semaphore(MAX_CONCURRENCY)

async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> str:
    async with sem:
        for attempt in range(1, MAX_RETRIES + 1):
            try:
                resp = await client.get(url, timeout=12.0)
                if resp.status_code == 429:
                    raise httpx.HTTPStatusError("Rate limit", request=resp.request, response=resp)
                resp.raise_for_status()
                soup = BeautifulSoup(resp.text, "lxml")
                return soup.title.string.strip() if soup.title else "No title"
            except (httpx.HTTPError, httpx.TimeoutException) as exc:
                if attempt == MAX_RETRIES:
                    return f"Failed: {exc}"
                backoff = BASE_DELAY * (2 ** (attempt - 1)) + random.random()
                await asyncio.sleep(backoff)

async def batch_fetch(urls: List[str]) -> dict:
    async with httpx.AsyncClient(http2=True) as client:
        tasks = [fetch_with_retry(client, u) for u in urls]
        results = await tqdm_asyncio.gather(*tasks, desc="Scraping", unit="url")
    return dict(zip(urls, results))

if __name__ == "__main__":
    # Example: 500 URLs from a public sitemap (trimmed for brevity)
    with open("sitemap_urls.txt") as f:
        url_list = [line.strip() for line in f if line.strip()]

    start = time.time()
    scraped = asyncio.run(batch_fetch(url_list))
    elapsed = time.time() - start
    print(f"\nFinished {len(url_list)} URLs in {elapsed:.2f}s ({len(url_list)/elapsed:.2f} req/s)")

    # Persist results
    import json
    with open("titles.json", "w") as out:
        json.dump(scraped, out, indent=2)

Key takeaways from this version:

• The semaphore caps concurrent connections to MAX_CONCURRENCY, keeping network usage predictable.
• Exponential back‑off (with a jitter component) prevents thundering‑herd retries that could trigger bans.
• tqdm_asyncio.gather gives you a live progress bar without extra boilerplate.

Real‑World Use Case: Competitive Price Monitoring

Imagine you run an e‑commerce analytics startup that tracks competitor pricing across dozens of retailer sites. Your pipeline must:

1. Collect product pages every hour.
2. Extract price, stock status, and promotional badges.
3. Store the data in a time‑series database for trend analysis.

Using the async scraper we just built, you can spin up a scheduled Azure Function or AWS Lambda (with a custom runtime) that pulls 1,000 URLs in under two minutes. The low memory footprint (< 150 MB) makes it cheap to run at scale.

Below is a snippet that extends the previous example to parse price information from a typical HTML snippet:

def extract_price(soup: BeautifulSoup) -> float | None:
    # Common patterns: $199.99 or 
    price_tag = soup.select_one("span.price, meta[itemprop='price']")
    if not price_tag:
        return None
    raw = price_tag.get_text() if price_tag.name == "span" else price_tag["content"]
    # Strip currency symbols and commas
    cleaned = raw.replace("$", "").replace(",", "").strip()
    try:
        return float(cleaned)
    except ValueError:
        return None

# Inside fetch_with_retry, replace the title return with a dict:
return {
    "title": soup.title.string.strip() if soup.title else "No title",
    "price": extract_price(soup),
    "url": url,
}

When you dump the final JSON, each entry now contains a structured payload ready for ingestion into InfluxDB or TimescaleDB. You can then visualize price trends with Grafana dashboards, alert on sudden spikes, or feed the data into a machine‑learning model that predicts competitor promotions.

Pro tip: Always respect the robots.txt file and include a custom User‑Agent header that identifies your service. Some sites block generic user agents like “Python‑urllib/3.12”. Example:

headers = {
    "User-Agent": "CodeyaanScraper/1.0 (+https://codeyaan.com/bot)"
}
client = httpx.AsyncClient(headers=headers, http2=True)

Testing & Debugging Asynchronous Code

Debugging async code can feel like chasing shadows, but a few practices make it manageable:

• Enable logging. Set httpx logger to DEBUG to see request/response details.
• Use anyio or trio for richer cancellation support.
• Write unit tests with pytest-asyncio. Mock httpx.AsyncClient using respx to simulate network conditions.

import pytest, respx, httpx, asyncio
from my_scraper import fetch_with_retry

@pytest.mark.asyncio
async def test_retry_on_429():
    url = "https://example.com/slow"
    with respx.mock(base_url="https://example.com") as mock:
        # First two attempts return 429, third succeeds
        mock.get("/slow").side_effect = [
            httpx.Response(429),
            httpx.Response(429),
            httpx.Response(200, text="<title>Success</title>")
        ]
        async with httpx.AsyncClient() as client:
            title = await fetch_with_retry(client, url)
            assert title == "Success"

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

CMD ["python", "-m", "my_scraper"]