Asynchronous Web Scraping in Python: Build Faster Scrapers With asyncio and aiohttp

TL;DR

• Async scraping in Python uses asyncio and aiohttp to send many requests at once on a single thread. A multi-URL scrape finishes in seconds instead of minutes.
• The baseline pattern combines ClientSession for connection pooling, asyncio.Semaphore for concurrency, split 5xx/429 backoff for retries, and JSON/CSV for export.
• At scale, add rotating residential proxies – and escalate to TLS impersonation (curl_cffi) or headless browsers (Patchright) when targets start blocking.
• For the hardest targets, a managed service like Decodo's Site Unblocker or Web Scraping API removes the anti-bot maintenance entirely.

What is asyncio and how does it work?

Python's asyncio is a standard-library module for writing concurrent code that runs on a single thread. Instead of creating operating-system threads, it uses an event loop that schedules coroutines, pausing one whenever it reaches an I/O wait and passing the CPU to another. For network-bound work like web scraping, that pause-and-switch pattern is what makes async faster.

Three concepts matter:

• Event loop – the scheduler. It watches which coroutines are ready to run and which are waiting on I/O, and interleaves them.
• Coroutine – a function defined with async def. Calling it returns a coroutine object; it doesn't run until awaited.
• Task – a coroutine scheduled on the event loop. asyncio.create_task() schedules one; asyncio.gather() schedules many and waits for all.

Here's a minimal example with no network calls, so you can see the interleaving:

import asyncio
import time

async def fetch_mock(name: str, delay: float) -> str:
    print(f"[{time.strftime('%X')}] start {name}")
    await asyncio.sleep(delay)  # simulates waiting on a response
    print(f"[{time.strftime('%X')}] done  {name}")
    return f"response from {name}"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch_mock("A", 2.0),
        fetch_mock("B", 3.0),
        fetch_mock("C", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(f"finished in {elapsed:.2f}s")
    print(results)

asyncio.run(main())

import asyncio
import time

async def fetch_mock(name: str, delay: float) -> str:
    print(f"[{time.strftime('%X')}] start {name}")
    await asyncio.sleep(delay)  # simulates waiting on a response
    print(f"[{time.strftime('%X')}] done  {name}")
    return f"response from {name}"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch_mock("A", 2.0),
        fetch_mock("B", 3.0),
        fetch_mock("C", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(f"finished in {elapsed:.2f}s")
    print(results)

asyncio.run(main())

Running it produces interleaved timestamps that show the overlap directly:

[12:00:00] start A
[12:00:00] start B
[12:00:00] start C
[12:00:01] done  C
[12:00:02] done  A
[12:00:03] done  B
finished in 3.00s

[12:00:00] start A
[12:00:00] start B
[12:00:00] start C
[12:00:01] done  C
[12:00:02] done  A
[12:00:03] done  B
finished in 3.00s

All 3 coroutines start in the same second, each pauses during its asyncio.sleep(), and the event loop resumes each when its timer expires – C (1s) finishes first, then A (2s), then B (3s). Total time is the slowest task, not the sum. Real HTTP requests behave the same way: the slow part is waiting for bytes, which asyncio overlaps across many tasks.

asyncio gives you 3 ways to wait on a batch:

• asyncio.gather(*tasks) – waits for every task and returns results in the same order as the inputs. The * in *tasks is Python's argument-unpacking operator; it expands a list into separate positional arguments, so gather(*[a, b, c]) is the same as gather(a, b, c). Use gather when you need the full batch before continuing.
• asyncio.as_completed(tasks) – yields tasks in the order they finish. Use it when you want to process each result as soon as it arrives, for example to write to disk without waiting for the slowest response.
• asyncio.TaskGroup() (Python 3.11+) – an async context manager. You create tasks with tg.create_task() inside the async with block, and the group applies structured concurrency: if any task raises an exception, every other task in the group is canceled, the block waits for cleanup to finish, then raises an ExceptionGroup that bundles the failures together. (ExceptionGroup is a Python 3.11 type for grouping multiple exceptions into one; you catch it with the new except* syntax – note the asterisk). The gather() function is still common for simple batch waits, but TaskGroup is the safer default when failures matter, because it doesn't leave tasks running after another task in the group has already crashed.

The rest of this guide uses gather() because it keeps the examples short and matches the 3.10+ baseline. If you're on 3.11 or later and want stricter failure semantics, replace any await asyncio.gather(*tasks) with a TaskGroup block; the rest of the code doesn't change.

asyncio isn't the same as threading or multiprocessing. Threads run multiple OS threads and work well for I/O-bound code that can't be rewritten in async form. Multiprocessing runs multiple Python processes and is typically the right choice for CPU-bound work like image processing. asyncio is the lightest of the three: one thread, one process, thousands of coroutines. 

For scraping, where nearly all the time is network I/O, asyncio handles more concurrent requests with the same amount of memory – a coroutine costs a few KB, while an OS thread reserves around 8 MB of stack on typical Linux defaults. The trade-offs between these models are covered in the guide on concurrency vs. parallelism.

Synchronous vs. asynchronous web scraping: performance compared

Synchronous web scraping sends one request at a time and waits for each response before starting the next. Asynchronous web scraping sends many requests concurrently on a single thread, overlapping the network waits – which is why an async scraper finishes a 50-URL batch in seconds where a sync one takes minutes. In sync scraping, every network wait runs one after another, adding up in real time. Async overlaps them inside the time of the slowest one:

To verify the speedup, build the same scraper twice – once with Requests, once with aiohttp – and run both against the same target. The example below fetches the Hacker News front page, extracts story links, then fetches each story page and parses out title, score, author, and comment count. Different HTTP clients suit different jobs; the guide on HTTPX vs. Requests vs. AIOHTTP covers when to pick which.

The synchronous baseline looks like this:

# sync_scraper.py
import time
import requests
from bs4 import BeautifulSoup

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    )
}
BASE = "https://news.ycombinator.com/"

def fetch(url: str) -> str:
    response = requests.get(url, headers=HEADERS, timeout=30)
    response.raise_for_status()
    return response.text

def extract_story_urls(html: str) -> list[str]:
    """Pull HN discussion URLs from the front page using each row's id attribute."""
    soup = BeautifulSoup(html, "html.parser")
    urls = []
    for row in soup.select("tr.athing"):
        story_id = row.get("id")
        if story_id:
            urls.append(f"{BASE}item?id={story_id}")
    return urls

def parse_story(html: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    title_tag = soup.select_one("span.titleline > a")
    score_tag = soup.select_one("span.score")
    author_tag = soup.select_one("a.hnuser")
    subtext = soup.select_one("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.select("a"):
            text = anchor.get_text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.get_text(strip=True) if title_tag else None,
        "url": title_tag.get("href") if title_tag else None,
        "score": int(score_tag.get_text().split()[0]) if score_tag else 0,
        "author": author_tag.get_text(strip=True) if author_tag else None,
        "comments": comments,
    }

def run():
    front_html = fetch(BASE)
    urls = extract_story_urls(front_html)

    start = time.perf_counter()
    results = [parse_story(fetch(url)) for url in urls]
    elapsed = time.perf_counter() - start

    print(f"sync: {len(results)} stories in {elapsed:.2f}s")
    return results

if __name__ == "__main__":
    run()

# sync_scraper.py
import time
import requests
from bs4 import BeautifulSoup

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    )
}
BASE = "https://news.ycombinator.com/"

def fetch(url: str) -> str:
    response = requests.get(url, headers=HEADERS, timeout=30)
    response.raise_for_status()
    return response.text

def extract_story_urls(html: str) -> list[str]:
    """Pull HN discussion URLs from the front page using each row's id attribute."""
    soup = BeautifulSoup(html, "html.parser")
    urls = []
    for row in soup.select("tr.athing"):
        story_id = row.get("id")
        if story_id:
            urls.append(f"{BASE}item?id={story_id}")
    return urls

def parse_story(html: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    title_tag = soup.select_one("span.titleline > a")
    score_tag = soup.select_one("span.score")
    author_tag = soup.select_one("a.hnuser")
    subtext = soup.select_one("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.select("a"):
            text = anchor.get_text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.get_text(strip=True) if title_tag else None,
        "url": title_tag.get("href") if title_tag else None,
        "score": int(score_tag.get_text().split()[0]) if score_tag else 0,
        "author": author_tag.get_text(strip=True) if author_tag else None,
        "comments": comments,
    }

def run():
    front_html = fetch(BASE)
    urls = extract_story_urls(front_html)

    start = time.perf_counter()
    results = [parse_story(fetch(url)) for url in urls]
    elapsed = time.perf_counter() - start

    print(f"sync: {len(results)} stories in {elapsed:.2f}s")
    return results

if __name__ == "__main__":
    run()

If you used the href on span.titleline > a, you would scrape the external article. The HN discussion URL has to be built from the row's id attribute (the story ID), which is why extract_story_urls reads row.get("id") and constructs item?id=… directly. To avoid this on your own targets, inspect the element in browser devtools and check the actual href / id before trusting a selector.

The async version reuses the same extract_story_urls and parse_story functions and only changes the fetch strategy. Save the file as async_scraper.py next to sync_scraper.py in the same directory so the import resolves:

# async_scraper.py
import asyncio
import time
import aiohttp
from sync_scraper import extract_story_urls, parse_story, HEADERS, BASE

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.text()

async def run():
    async with aiohttp.ClientSession(headers=HEADERS) as session:
        front_html = await fetch(session, BASE)
        urls = extract_story_urls(front_html)

        start = time.perf_counter()
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
        results = [parse_story(p) for p in pages]
        elapsed = time.perf_counter() - start

    print(f"async: {len(results)} stories in {elapsed:.2f}s")
    return results

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

# async_scraper.py
import asyncio
import time
import aiohttp
from sync_scraper import extract_story_urls, parse_story, HEADERS, BASE

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.text()

async def run():
    async with aiohttp.ClientSession(headers=HEADERS) as session:
        front_html = await fetch(session, BASE)
        urls = extract_story_urls(front_html)

        start = time.perf_counter()
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
        results = [parse_story(p) for p in pages]
        elapsed = time.perf_counter() - start

    print(f"async: {len(results)} stories in {elapsed:.2f}s")
    return results

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

To measure the speedup without sending heavy load to a real target (or being rate-limited during the benchmark), the cleanest demo is to point both scrapers at httpbin.org/delay/1, an endpoint that waits one second before responding. Each request has roughly one second of network wait, so the sync scraper pays it N times and the async scraper overlaps them.

# benchmark.py
import asyncio
import time
import requests
import aiohttp

URL = "https://httpbin.org/delay/1"
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    )
}

def run_sync(n: int) -> float:
    start = time.perf_counter()
    for _ in range(n):
        requests.get(URL, headers=HEADERS, timeout=30)
    return time.perf_counter() - start

async def run_async(n: int, concurrency: int) -> float:
    sem = asyncio.Semaphore(concurrency)
    async def bound(s):
        async with sem:
            async with s.get(URL) as r:
                return r.status
    timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as s:
        start = time.perf_counter()
        await asyncio.gather(*[bound(s) for _ in range(n)])
        return time.perf_counter() - start

if __name__ == "__main__":
    print(f"{'count':<8}{'sync (s)':<14}{'async (s)':<14}{'speedup':<10}")
    print("-" * 50)
    for n in [10, 25, 50]:
        sync_t = run_sync(n)
        async_t = asyncio.run(run_async(n, concurrency=20))
        print(f"{n:<8}{sync_t:<14.2f}{async_t:<14.2f}{sync_t/async_t:.1f}x")

# benchmark.py
import asyncio
import time
import requests
import aiohttp

URL = "https://httpbin.org/delay/1"
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    )
}

def run_sync(n: int) -> float:
    start = time.perf_counter()
    for _ in range(n):
        requests.get(URL, headers=HEADERS, timeout=30)
    return time.perf_counter() - start

async def run_async(n: int, concurrency: int) -> float:
    sem = asyncio.Semaphore(concurrency)
    async def bound(s):
        async with sem:
            async with s.get(URL) as r:
                return r.status
    timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as s:
        start = time.perf_counter()
        await asyncio.gather(*[bound(s) for _ in range(n)])
        return time.perf_counter() - start

if __name__ == "__main__":
    print(f"{'count':<8}{'sync (s)':<14}{'async (s)':<14}{'speedup':<10}")
    print("-" * 50)
    for n in [10, 25, 50]:
        sync_t = run_sync(n)
        async_t = asyncio.run(run_async(n, concurrency=20))
        print(f"{n:<8}{sync_t:<14.2f}{async_t:<14.2f}{sync_t/async_t:.1f}x")

Output from a real run on a residential connection (your numbers will vary with network round-trip time and httpbin's load, but the ratios stay in the same range):

Each request to httpbin.org/delay/1 costs 1 second of server delay plus 2-3 seconds of network round-trip and httpbin's variable load, so the per-request wall time on a residential link is closer to 3-4 seconds than to 1. The async time stays close to flat as N grows until the concurrency cap is fully used, which is why the speedup widens with batch size.

The same pattern holds against real-world targets (HN, news listings, product pages), but absolute numbers depend on the target's response time, your network latency, and how strictly the target rate-limits a single IP. A high-volume scraper reaches the rate-limit before reaching the throughput maximum, which is why proxies (covered later in this guide) matter.

Sync scraping isn't always worse. It's the better choice when:

• You need to scrape a single page, where the setup cost of asyncio doesn't pay off.
• You're debugging selectors and want a fast print-then-inspect cycle.
• The target requires sequential interaction (login, then form, then result) where request order matters and there's no parallelism to exploit.

For anything past a few dozen pages against a tolerant target, async is the right default.

Setting up your async web scraping project

A clean project layout and a proxy slot set up from the start will save an hour of refactoring later. Proxies matter because a high request rate from a single IP often gets blocked quickly on strict targets, and adding proxy auth to a scraper afterwards usually means modifying every fetch call; the article on residential proxies explains why rotating residential IPs help at scale.

The stack is Python 3.10 or newer, aiohttp on the 3.13.x line, beautifulsoup4 4.14 or later, and python-dotenv for credentials. Why Python 3.10 as the minimum: the str | None union syntax used in the type hints below requires it, and the optional escalation tools (curl_cffi 0.14+ and Scrapling) both require 3.10+. Python 3.11 or later also enables asyncio.TaskGroup and the asyncio.timeout() context manager; 3.14 adds the python -m asyncio ps introspection CLI covered later.

If you have uv installed – the Rust-based package manager from Astral that became one of the most-adopted Python install tools in 2025-2026, 10-100x faster than pip on most workloads – the setup is 2 commands inside a fresh project directory:

mkdir async-scraper && cd async-scraper
uv venv
source .venv/bin/activate          # macOS / Linux
# .venv\Scripts\activate           # Windows

uv pip install "aiohttp>=3.13" "beautifulsoup4>=4.14" python-dotenv

mkdir async-scraper && cd async-scraper
uv venv
source .venv/bin/activate          # macOS / Linux
# .venv\Scripts\activate           # Windows

uv pip install "aiohttp>=3.13" "beautifulsoup4>=4.14" python-dotenv

If you use pip, the equivalent works the same, just slower:

python -m venv .venv
source .venv/bin/activate

pip install "aiohttp>=3.13" "beautifulsoup4>=4.14" python-dotenv

python -m venv .venv
source .venv/bin/activate

pip install "aiohttp>=3.13" "beautifulsoup4>=4.14" python-dotenv

For one-off scripts where setting up a project is more than you need, uv also supports PEP 723 inline script metadata: declare the dependencies inside a comment block at the top of a single .py file and run it with uv run, which builds a temporary environment automatically when you run the script.

# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "aiohttp>=3.13",
#     "beautifulsoup4>=4.14",
# ]
# ///
import asyncio, aiohttp
# ... your scraper here ...

# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "aiohttp>=3.13",
#     "beautifulsoup4>=4.14",
# ]
# ///
import asyncio, aiohttp
# ... your scraper here ...

uv run scraper.py     # builds env from the inline block, runs the script

uv run scraper.py     # builds env from the inline block, runs the script

This is useful for sharing a working scraper as one file: the recipient runs uv run scraper.py and the dependencies install automatically. The full multi-file project layout below is still the right structure for anything you'll maintain long-term.

aiohttp's stable releases don't support HTTP/2 (open issue upstream but not yet released); if you're scraping targets that serve measurably different content or prioritize HTTP/2 clients, HTTPX is the common alternative (async API, HTTP/2 support, slower at raw throughput). For most scraping work where the target serves plain HTTP/1.1 HTML, aiohttp's throughput advantage makes it the better choice.

Lay out the project with one file per responsibility. Small files are easier to test and change than a single scraper script that does everything:

async-scraper/
├── .env                 # proxy credentials (gitignored)
├── config.py            # credentials, headers, concurrency settings
├── parser.py            # Beautiful Soup extraction logic
├── resilient_fetch.py   # retry / backoff wrapper
├── export.py            # JSON and CSV writers
└── scraper.py           # the orchestrator that ties them together

async-scraper/
├── .env                 # proxy credentials (gitignored)
├── config.py            # credentials, headers, concurrency settings
├── parser.py            # Beautiful Soup extraction logic
├── resilient_fetch.py   # retry / backoff wrapper
├── export.py            # JSON and CSV writers
└── scraper.py           # the orchestrator that ties them together

Store your Decodo residential proxy credentials in a .env file at the project root. Keeping them out of code means the repo stays safe to commit, and rotating credentials is a one-line change. The username and password come from your Decodo dashboard once your residential plan is active; the same gateway host and port work for rotating, sticky, and geo-targeted requests, with the differences encoded in the username string. 

If you don't have proxy credentials yet, leave DECODO_USER and DECODO_PASS unset – PROXY_URL will default to None in config.py and the rest of this guide will run against a tolerant target without a proxy. Add real credentials before increasing volume against any site that enforces rate limits.

# .env
DECODO_USER=your_username
DECODO_PASS=your_password
DECODO_HOST=gate.decodo.com
DECODO_PORT=7000

# .env
DECODO_USER=your_username
DECODO_PASS=your_password
DECODO_HOST=gate.decodo.com
DECODO_PORT=7000

Load those values in config.py so every module reads them from a single place:

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

DECODO_USER = os.getenv("DECODO_USER")
DECODO_PASS = os.getenv("DECODO_PASS")
DECODO_HOST = os.getenv("DECODO_HOST", "gate.decodo.com")
DECODO_PORT = os.getenv("DECODO_PORT", "7000")

PROXY_URL = (
    f"http://{DECODO_USER}:{DECODO_PASS}@{DECODO_HOST}:{DECODO_PORT}"
    if DECODO_USER and DECODO_PASS
    else None
)

DEFAULT_HEADERS = {
    # Realistic Chrome header set - the practical default for general scraping
    # in 2026. Update the Chrome version periodically (check chromium.org/release):
    # anti-bot vendors flag UAs more than 6-9 months stale.
    #
    # For cooperative targets where you want to identify your scraper - sites
    # you own, public-data sources, partners with explicit allowance - swap
    # the User-Agent for a polite form like:
    #   "Mozilla/5.0 (compatible; YourProject/1.0; +https://your-contact-url)"
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    ),
    "Accept": (
        "text/html,application/xhtml+xml,application/xml;q=0.9,"
        "image/avif,image/webp,image/apng,*/*;q=0.8,"
        "application/signed-exchange;v=b3;q=0.7"
    ),
    "Accept-Language": "en-US,en;q=0.9",
    # "br" (brotli) needs brotli or brotlicffi installed to decode.
    "Accept-Encoding": "gzip, deflate, br",
    "Sec-Ch-Ua": '"Google Chrome";v="142", "Not-A.Brand";v="99"',
    "Sec-Ch-Ua-Mobile": "?0",
    "Sec-Ch-Ua-Platform": '"macOS"',
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Upgrade-Insecure-Requests": "1",
}

CONCURRENCY = 10
REQUEST_TIMEOUT = 30

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

DECODO_USER = os.getenv("DECODO_USER")
DECODO_PASS = os.getenv("DECODO_PASS")
DECODO_HOST = os.getenv("DECODO_HOST", "gate.decodo.com")
DECODO_PORT = os.getenv("DECODO_PORT", "7000")

PROXY_URL = (
    f"http://{DECODO_USER}:{DECODO_PASS}@{DECODO_HOST}:{DECODO_PORT}"
    if DECODO_USER and DECODO_PASS
    else None
)

DEFAULT_HEADERS = {
    # Realistic Chrome header set - the practical default for general scraping
    # in 2026. Update the Chrome version periodically (check chromium.org/release):
    # anti-bot vendors flag UAs more than 6-9 months stale.
    #
    # For cooperative targets where you want to identify your scraper - sites
    # you own, public-data sources, partners with explicit allowance - swap
    # the User-Agent for a polite form like:
    #   "Mozilla/5.0 (compatible; YourProject/1.0; +https://your-contact-url)"
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/142.0.0.0 Safari/537.36"
    ),
    "Accept": (
        "text/html,application/xhtml+xml,application/xml;q=0.9,"
        "image/avif,image/webp,image/apng,*/*;q=0.8,"
        "application/signed-exchange;v=b3;q=0.7"
    ),
    "Accept-Language": "en-US,en;q=0.9",
    # "br" (brotli) needs brotli or brotlicffi installed to decode.
    "Accept-Encoding": "gzip, deflate, br",
    "Sec-Ch-Ua": '"Google Chrome";v="142", "Not-A.Brand";v="99"',
    "Sec-Ch-Ua-Mobile": "?0",
    "Sec-Ch-Ua-Platform": '"macOS"',
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Upgrade-Insecure-Requests": "1",
}

CONCURRENCY = 10
REQUEST_TIMEOUT = 30

That's the practical default for general scraping in 2026. The Sec-Ch-Ua and Sec-Fetch-* headers are sent by every modern Chrome and their absence is a strong "this is automation" signal, so they belong in the default set even when a target doesn't strictly check them today. 

The polite alternative in the comment is the right choice for cooperative targets where the relationship matters more than looking like a browser. Pair this header set with curl_cffi's TLS impersonation later in this guide for a stack that matches Chrome at both the TLS layer and the HTTP-header layer.

Before you write any extraction logic, confirm aiohttp is installed correctly and the network is working. Save the script below as verify.py in the project root, then run python verify.py (with the virtual environment activated). Expect a 200 status and a JSON body echoing your User-Agent. If you see ModuleNotFoundError: No module named 'aiohttp', your virtual environment isn't active – re-run source .venv/bin/activate.

# verify.py
import asyncio
import aiohttp
from config import DEFAULT_HEADERS

async def main():
    async with aiohttp.ClientSession(headers=DEFAULT_HEADERS) as session:
        async with session.get("https://httpbin.org/get") as resp:
            print(resp.status, await resp.json())

asyncio.run(main())

# verify.py
import asyncio
import aiohttp
from config import DEFAULT_HEADERS

async def main():
    async with aiohttp.ClientSession(headers=DEFAULT_HEADERS) as session:
        async with session.get("https://httpbin.org/get") as resp:
            print(resp.status, await resp.json())

asyncio.run(main())

If this fails, fix networking or dependency issues before continuing; debugging is much harder once retries and proxies are added on top.

Is uvloop still faster than asyncio?

On Linux and macOS, uvloop is a drop-in asyncio event loop built on libuv, the same C library that Node.js uses for its event loop. It used to deliver a 2x to 4x throughput improvement on network-heavy workloads. On Python 3.13 and 3.14 the difference has shrunk for HTTP scraping – CPython's native asyncio loop has improved enough that the difference varies by workload, sometimes in either direction. Treat uvloop as a measure-then-decide optimization, not an automatic default.

Where uvloop still helps: code that does lots of small awaits in tight loops, timer-driven workloads, and WebSocket servers that broadcast to many clients at once.

uv pip install uvloop

uv pip install uvloop

The current API is uvloop.run(), a drop-in replacement for asyncio.run() (the older uvloop.install() relies on asyncio.set_event_loop_policy(), deprecated in Python 3.14 and slated for removal in 3.16). On Windows, use asyncio.run() instead since uvloop doesn't run there.

Sending async HTTP requests with aiohttp

The async benchmark above used aiohttp.ClientSession in its simplest form, but the most common performance bug in async scrapers comes from a misunderstanding of what it is. The ClientSession class is a connection pool, not a convenience wrapper around session.get(). It holds open TCP and TLS connections to the same host so the next request can reuse them – what HTTP calls keep-alive. (Each new HTTPS request normally costs an extra round-trip to set up the TCP connection, plus a few more for the TLS encryption handshake; keep-alive lets the next request skip both). Readers used to sync requests can compare the patterns in the guide on Python Requests.

Creating 1 session per request discards the pool, forces a new TCP connection and TLS handshake every time, and slows the scraper down to something close to sync speed. The correct pattern is 1 session for the whole run:

# session_example.py
import asyncio
import aiohttp
from config import DEFAULT_HEADERS, REQUEST_TIMEOUT

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.text()

async def main():
    urls = [
        "https://news.ycombinator.com/",
        "https://news.ycombinator.com/newest",
        "https://news.ycombinator.com/ask",
    ]
    timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
    async with aiohttp.ClientSession(
        headers=DEFAULT_HEADERS,
        timeout=timeout,
        raise_for_status=True,
    ) as session:
        tasks = [fetch(session, u) for u in urls]
        pages = await asyncio.gather(*tasks)
        for url, page in zip(urls, pages):
            print(url, len(page), "bytes")

asyncio.run(main())

# session_example.py
import asyncio
import aiohttp
from config import DEFAULT_HEADERS, REQUEST_TIMEOUT

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.text()

async def main():
    urls = [
        "https://news.ycombinator.com/",
        "https://news.ycombinator.com/newest",
        "https://news.ycombinator.com/ask",
    ]
    timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
    async with aiohttp.ClientSession(
        headers=DEFAULT_HEADERS,
        timeout=timeout,
        raise_for_status=True,
    ) as session:
        tasks = [fetch(session, u) for u in urls]
        pages = await asyncio.gather(*tasks)
        for url, page in zip(urls, pages):
            print(url, len(page), "bytes")

asyncio.run(main())

A few details in this snippet matter for production scrapers:

• headers=DEFAULT_HEADERS on the session applies to every request inside the async with block. Set the User-Agent once there rather than on every session.get() call.
• timeout=ClientTimeout(total=30, connect=5, sock_read=10) on the session sets a default timeout for every request without you having to pass one to each session.get(). The aiohttp default is total=300 with no other per-phase caps, which is too generous for most scrapers. Splitting the budget across connect (max time to open the connection) and sock_read (max time between 2 chunks of bytes) lets a stuck connect or a stalled read fail in seconds rather than waiting for the full total.
• raise_for_status=True on the session calls response.raise_for_status() automatically on every response, so 4xx and 5xx codes turn into ClientResponseError exceptions you can catch centrally.

One security note on raise_for_status: by default the exception's message attribute can include the original Authorization header value if the server echoes it back. If you're using basic auth on the proxy or target, either set raise_for_status=False and check status manually, or strip the header from logs before recording the exception.

If the target requires cookies, set them on the session so every request sends them automatically:

cookies = {"session_id": "abc123", "consent": "accepted"}
async with aiohttp.ClientSession(headers=DEFAULT_HEADERS, cookies=cookies) as session:
    ...

cookies = {"session_id": "abc123", "consent": "accepted"}
async with aiohttp.ClientSession(headers=DEFAULT_HEADERS, cookies=cookies) as session:
    ...

Custom headers per request (for example, a referer that changes per URL) go on the individual session.get() call and override the session defaults for that request only.

Parsing HTML and extracting data with Beautiful Soup

aiohttp returns HTML as a string; Beautiful Soup is the parser this guide uses for the reason most scraping stacks default to it: forgiving with malformed HTML, clean CSS-selector API, no extra binary dependencies. Keep parsing in its own function, separate from fetching, so you can save a few HTML files and iterate on selectors without making any network requests. More patterns for common HTML structures are in the guide on Beautiful Soup web scraping, and the trade-offs between selector styles are covered in XPath vs. CSS selectors.

The Hacker News story page has a predictable structure: one tr.athing row per story on the front page, and each story page has a span.titleline, a span.score, an a.hnuser, and a comment count inside td.subtext. The parser below handles all fields and returns None or 0 when a field is missing rather than crashing:

# parser.py
from bs4 import BeautifulSoup

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

    title_tag = soup.select_one("span.titleline > a")
    score_tag = soup.select_one("span.score")
    author_tag = soup.select_one("a.hnuser")
    subtext = soup.select_one("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.select("a"):
            text = anchor.get_text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.get_text(strip=True) if title_tag else None,
        "url": title_tag.get("href") if title_tag else None,
        "score": int(score_tag.get_text().split()[0]) if score_tag else 0,
        "author": author_tag.get_text(strip=True) if author_tag else None,
        "comments": comments,
    }

# parser.py
from bs4 import BeautifulSoup

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

    title_tag = soup.select_one("span.titleline > a")
    score_tag = soup.select_one("span.score")
    author_tag = soup.select_one("a.hnuser")
    subtext = soup.select_one("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.select("a"):
            text = anchor.get_text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.get_text(strip=True) if title_tag else None,
        "url": title_tag.get("href") if title_tag else None,
        "score": int(score_tag.get_text().split()[0]) if score_tag else 0,
        "author": author_tag.get_text(strip=True) if author_tag else None,
        "comments": comments,
    }

Returning a dictionary (rather than a tuple or a custom class) keeps the parser easy to use with JSON and CSV exporters. Defensive checks on every select_one result matter because HN occasionally serves a deleted or flagged story with fewer fields; crashing on None.get_text() in the middle of a 500-URL run wastes every successful response that came before it.

One note on the html.parser backend: it's part of the Python standard library, so you don't need lxml or html5lib installed. It's a bit slower and more forgiving of malformed HTML than lxml. For large pages or limited CPU, switch to lxml and the only change is the second argument to BeautifulSoup().

selectolax vs. Beautiful Soup: when parsing becomes the bottleneck

At a small scale, Beautiful Soup parsing time is small compared to network time. At a large scale, parsing can take longer than fetching. selectolax is a Python wrapper around the Lexbor HTML parser (written in C) and is significantly faster than both Beautiful Soup backends. Running 500 parses of the same ~700 KB HN story page on Python 3.14:

selectolax is also 31.5x faster than Beautiful Soup with lxml, which is the fastest Beautiful Soup configuration. On a 500-URL scrape, that's the difference between 50 seconds of parsing and 1.6 seconds. v0.4.7 is the current release, and the underlying lexbor backend is actively developed and more spec-compliant than the older, now-deprecated modest backend.

uv pip install selectolax

uv pip install selectolax

Here’s the code:

# parser_fast.py
from selectolax.lexbor import LexborHTMLParser

def parse_story_fast(html: str) -> dict:
    tree = LexborHTMLParser(html)
    title_tag = tree.css_first("span.titleline > a")
    score_tag = tree.css_first("span.score")
    author_tag = tree.css_first("a.hnuser")
    subtext = tree.css_first("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.css("a"):
            text = anchor.text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.text(strip=True) if title_tag else None,
        "url": title_tag.attributes.get("href") if title_tag else None,
        "score": int(score_tag.text().split()[0]) if score_tag else 0,
        "author": author_tag.text(strip=True) if author_tag else None,
        "comments": comments,
    }

# parser_fast.py
from selectolax.lexbor import LexborHTMLParser

def parse_story_fast(html: str) -> dict:
    tree = LexborHTMLParser(html)
    title_tag = tree.css_first("span.titleline > a")
    score_tag = tree.css_first("span.score")
    author_tag = tree.css_first("a.hnuser")
    subtext = tree.css_first("td.subtext")

    comments = 0
    if subtext:
        for anchor in subtext.css("a"):
            text = anchor.text(strip=True)
            if "comment" in text:
                parts = text.split()
                if parts and parts[0].isdigit():
                    comments = int(parts[0])

    return {
        "title": title_tag.text(strip=True) if title_tag else None,
        "url": title_tag.attributes.get("href") if title_tag else None,
        "score": int(score_tag.text().split()[0]) if score_tag else 0,
        "author": author_tag.text(strip=True) if author_tag else None,
        "comments": comments,
    }

The API is close enough to Beautiful Soup that porting a parser is straightforward: select_one becomes css_first, select becomes css, get_text becomes text, and attribute access uses .attributes. The trade-off is that selectolax is stricter about malformed HTML; pages with deeply broken markup still parse better with Beautiful Soup. A common pattern: try selectolax first, use Beautiful Soup as a backup if a required selector returns None.

Controlling concurrency with asyncio.Semaphore

To limit concurrent requests in asyncio, wrap each fetch task in an asyncio.Semaphore(N) – only N tasks can hold the semaphore at once, the rest wait. This is the standard pattern for capping concurrency in aiohttp scrapers, and it solves the hidden problem in asyncio.gather(): gather does not limit how many tasks run at once. Pass 500 tasks and aiohttp tries to open 500 connections, which can overwhelm the target, trigger rate limiting or IP bans, and sometimes reach your local OS file-descriptor limit (default 1024 on Linux) before the requests even reach the server. Broader anti-detection techniques are covered in the guide on anti-scraping techniques.

Wrap the fetch inside an async with on the semaphore and you get a configurable concurrency cap with about 3 extra lines of code:

# throttled_fetch.py
import asyncio
import random
import aiohttp
from config import DEFAULT_HEADERS, REQUEST_TIMEOUT, CONCURRENCY

async def throttled_fetch(
    semaphore: asyncio.Semaphore,
    session: aiohttp.ClientSession,
    url: str,
) -> str:
    async with semaphore:
        async with session.get(url) as response:
            response.raise_for_status()
            text = await response.text()
        # small randomized jitter between batches lowers the detection signature
        await asyncio.sleep(random.uniform(0.1, 0.5))
        return text

async def scrape(urls: list[str]) -> list[str]:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT, connect=5, sock_read=10)
    async with aiohttp.ClientSession(headers=DEFAULT_HEADERS, timeout=timeout) as session:
        tasks = [throttled_fetch(semaphore, session, u) for u in urls]
        return await asyncio.gather(*tasks)

# throttled_fetch.py
import asyncio
import random
import aiohttp
from config import DEFAULT_HEADERS, REQUEST_TIMEOUT, CONCURRENCY

async def throttled_fetch(
    semaphore: asyncio.Semaphore,
    session: aiohttp.ClientSession,
    url: str,
) -> str:
    async with semaphore:
        async with session.get(url) as response:
            response.raise_for_status()
            text = await response.text()
        # small randomized jitter between batches lowers the detection signature
        await asyncio.sleep(random.uniform(0.1, 0.5))
        return text

async def scrape(urls: list[str]) -> list[str]:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT, connect=5, sock_read=10)
    async with aiohttp.ClientSession(headers=DEFAULT_HEADERS, timeout=timeout) as session:
        tasks = [throttled_fetch(semaphore, session, u) for u in urls]
        return await asyncio.gather(*tasks)

The right value for CONCURRENCY depends on three things: how tolerant the target is, how large your proxy pool is, and how fast each request responds. A safe starting point for most public targets is 10 to 20; increase only if the target accepts it and responses remain successful. A pool of rotating residential IPs lets you run higher concurrency because each request appears to come from a different client. (For cooperative targets where IP reputation isn't a factor, datacenter or ISP proxies are cheaper per GB; residential is the safer default when targets actively block).

Two throttles exist at different layers and they work together cleanly. asyncio.Semaphore caps coroutine-level concurrency: how many fetch tasks are allowed to run at once. aiohttp's TCPConnector caps connection-level concurrency: how many TCP sockets the pool will open. The connector defaults to limit=100 total and limit_per_host=0 (unlimited per host), which is too permissive for most scraping. Passing an explicit connector narrows both:

import aiohttp

connector = aiohttp.TCPConnector(limit=50, limit_per_host=10)
async with aiohttp.ClientSession(connector=connector, headers=DEFAULT_HEADERS) as session:
    ...