Python Asyncio Masterclass: Building a High-Concurrency Web Scraper with aiohttp

Why Asyncio Outperforms Threading for I/O-Bound Work

Python's GIL prevents true CPU parallelism in threads. But for I/O-bound tasks like HTTP requests, database queries, and file reads, the GIL is irrelevant — threads spend most of their time waiting, not executing Python bytecode. Asyncio takes this further: instead of spinning up one thread per connection (expensive — each thread costs ~8MB of stack memory), the event loop multiplexes thousands of concurrent I/O operations on a single OS thread using non-blocking epoll/kqueue system calls.

1. The Event Loop — How Asyncio Works

The asyncio event loop is a single-threaded scheduler. When a coroutine hits an await expression on an I/O operation, it suspends and yields control back to the loop. The loop picks up another coroutine and runs it until it also awaits. No OS context switches needed — just Python frame switching.

import asyncio

async def fetch_data(name: str, delay: float) -> str:
    print(f"[{name}] Starting...")
    await asyncio.sleep(delay)  # Simulates I/O wait
    print(f"[{name}] Done after {delay}s")
    return f"data from {name}"

async def main():
    # Run 3 coroutines CONCURRENTLY — total time ~1s, not 2.3s
    results = await asyncio.gather(
        fetch_data("API-1", 1.0),
        fetch_data("API-2", 0.8),
        fetch_data("API-3", 0.5),
    )
    print(results)

asyncio.run(main())
# [API-3] Done after 0.5s
# [API-2] Done after 0.8s
# [API-1] Done after 1.0s

2. Coroutines vs Tasks vs Futures

import asyncio

async def my_coroutine():
    return 42

# Coroutine object — NOT yet scheduled
coro = my_coroutine()

# Task — scheduled to run on the event loop immediately
task = asyncio.create_task(my_coroutine())

async def demonstrate():
    # SEQUENTIAL — each awaits before the next starts
    r1 = await my_coroutine()
    r2 = await my_coroutine()

    # CONCURRENT — both scheduled immediately
    t1 = asyncio.create_task(my_coroutine())
    t2 = asyncio.create_task(my_coroutine())
    r1, r2 = await asyncio.gather(t1, t2)

3. Production Async Web Scraper

import asyncio
import aiohttp
import json
import logging
from datetime import datetime
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AsyncScraper:
    def __init__(self, concurrency: int = 50, timeout: int = 30, max_retries: int = 3):
        self.semaphore = asyncio.Semaphore(concurrency)
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.max_retries = max_retries

    async def fetch_one(
        self, session: aiohttp.ClientSession, url: str, attempt: int = 1
    ) -> Optional[dict]:
        async with self.semaphore:  # Max N concurrent requests
            try:
                async with session.get(url, timeout=self.timeout) as resp:
                    text = await resp.text()
                    return {
                        "url": url,
                        "status": resp.status,
                        "length": len(text),
                        "scraped_at": datetime.utcnow().isoformat(),
                    }
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt < self.max_retries:
                    wait = 2 ** attempt  # Exponential backoff: 2s, 4s, 8s
                    logger.warning(f"Retry {attempt}/{self.max_retries} for {url}")
                    await asyncio.sleep(wait)
                    return await self.fetch_one(session, url, attempt + 1)
                return {"url": url, "status": None, "error": str(e)}

    async def scrape_all(self, urls: list[str]) -> list[dict]:
        connector = aiohttp.TCPConnector(limit=100, enable_cleanup_closed=True)
        headers = {"User-Agent": "AsyncScraper/1.0"}
        async with aiohttp.ClientSession(connector=connector, headers=headers) as session:
            tasks = [self.fetch_one(session, url) for url in urls]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return [r for r in results if r is not None]


async def main():
    urls = [f"https://httpbin.org/delay/1?n={i}" for i in range(100)]
    scraper = AsyncScraper(concurrency=20, timeout=15)
    results = await scraper.scrape_all(urls)
    print(f"Scraped {len(results)} URLs")
    print(f"Success: {sum(1 for r in results if r.get('status') == 200)}/{len(results)}")

asyncio.run(main())

4. Asyncio with Django — The Right Way

from django.http import JsonResponse
from asgiref.sync import sync_to_async
import aiohttp, asyncio

get_active_jobs = sync_to_async(
    lambda: list(JobPosting.objects.filter(status='ACTIVE').values('id', 'title'))
)

async def fetch_external_salary(job_title: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.salaries.dev/?q={job_title}") as resp:
            return await resp.json()

async def jobs_with_salary_view(request):
    jobs, salary_data = await asyncio.gather(
        get_active_jobs(),
        fetch_external_salary("Software Engineer"),
    )
    return JsonResponse({"jobs": jobs, "market_salary": salary_data})

Warning: Django's ORM is synchronous. Never call QuerySet.all() directly inside an async def view — always wrap it with sync_to_async. Django 5.x is adding native async ORM support.