Scraping asincrono in Python: aiohttp e asyncio

Quando bisogna scaricare migliaia o decine di migliaia di pagine, il codice sincrono va a sbattere contro un muro: ogni richiesta aspetta la risposta prima che parta la successiva. I thread aiutano, ma consumano memoria e si portano dietro un overhead notevole. L’asincronia risolve il problema con più eleganza: un solo thread mantiene migliaia di connessioni simultanee, passando dall’una all’altra mentre aspettano la rete.

Questo articolo è la continuazione avanzata della guida generale «Web scraping con Python». Le tecniche di base (librerie, codifiche) sono spiegate lì; qui vediamo come scalare lo scraping con asyncio.

Indice

1. Perché async accelera lo scraping
2. Come scarichiamo la pagina: aiohttp
3. Parsing del contenuto nel codice async
4. Accenti e codifiche nello scraping asincrono
5. Controllo della concorrenza: semafori
6. Proxy
7. Scraping attraverso TOR
8. HTTPS/SSL
9. Lavorare con i cookie
10. Stato della risposta e header
11. Code: asyncio.Queue
12. httpx come alternativa
13. Pro e contro

1. Perché async accelera lo scraping

Lo scraping è un’attività I/O-bound: per il 99% del tempo il programma non fa che aspettare la risposta del server. Nel codice sincrono quell’attesa va sprecata. L’asincronia permette, mentre una richiesta aspetta, di lanciarne centinaia di altre.

• Sincrono: 1000 pagine da 0,5 s l’una = ~500 secondi.
• Asincrono (100 alla volta): le stesse 1000 pagine = ~5 secondi.

A differenza dei thread, le coroutine non costano quasi nulla in memoria: decine di migliaia di task simultanei in un solo thread sono del tutto realistici. Confrontalo con l’approccio multithread della guida generale: async scala parecchio più in alto.

2. Come scarichiamo la pagina: aiohttp

aiohttp è il client HTTP asincrono standard. Il principio chiave: un unico ClientSession per tutto il programma (riutilizza le connessioni) e molte richieste simultanee tramite asyncio.gather.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
        return await resp.text()

async def main(urls):
    async with aiohttp.ClientSession(headers={"User-Agent": "MyBot/1.0"}) as session:
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks, return_exceptions=True)
        return pages

urls = [f"https://example.com/page/{i}" for i in range(1, 1001)]
results = asyncio.run(main(urls))

return_exceptions=True è importante: un task fallito non fa crollare l’intero gather, ma torna come oggetto eccezione, che potrai poi gestire.

3. Parsing del contenuto nel codice async

Una sfumatura importante: l’analisi dell’HTML in sé (BeautifulSoup, lxml) è un’operazione CPU sincrona. Se l’HTML è pesante, il parsing blocca l’event loop e azzera il guadagno dell’async. Le pagine leggere si possono analizzare direttamente nella coroutine:

from bs4 import BeautifulSoup

async def fetch_and_parse(session, url):
    async with session.get(url) as resp:
        html = await resp.text()
    soup = BeautifulSoup(html, "lxml")     # per le pagine leggere va bene
    return soup.find("h1").get_text(strip=True)

Se l’analisi è pesante, spostala in un pool di processi per non bloccare il loop:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def heavy_parse(html):
    soup = BeautifulSoup(html, "lxml")
    return [a["href"] for a in soup.select("a")]

async def fetch_and_parse(session, url, pool):
    async with session.get(url) as resp:
        html = await resp.text()
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(pool, heavy_parse, html)

Il dettaglio sui parser è nella guida generale e nell’articolo su lxml (una delle opzioni più veloci per i carichi async).

Spesso in modo asincrono non si interrogano pagine HTML ma API: il corpo della risposta è già strutturato e, al posto di un parser, basta await resp.json(). È più veloce e più affidabile che smontare il markup; le tecniche per lavorare con queste risposte sono raccolte in «Parsing di JSON».

4. Accenti e codifiche nello scraping asincrono

Con await resp.text(), aiohttp prova a dedurre la codifica dagli header. Sui siti datati che servono ancora ISO-8859-1 o Windows-1252 (proprio dove vivono le lettere accentate), il rilevamento a volte sbaglia e il testo arriva illeggibile. Le soluzioni sono le stesse del codice sincrono:

# opzione 1: codifica esplicita
html = await resp.text(encoding="utf-8")

# opzione 2: lavorare con i byte e passarli al parser
raw = await resp.read()
soup = BeautifulSoup(raw, "lxml")    # il parser leggerà da sé <meta charset>

# opzione 3: decodifica manuale
html = raw.decode("windows-1252", errors="replace")

La teoria completa del problema è nella guida generale, sezione «Codifiche».

5. Controllo della concorrenza: semafori

Lanciare 10.000 richieste in un colpo solo significa «stendere» il server e la propria rete, oltre a guadagnarsi un ban. La concorrenza si limita con un semaforo:

import asyncio
import aiohttp

async def fetch(session, url, semaphore):
    async with semaphore:                      # non più di N alla volta
        async with session.get(url) as resp:
            return await resp.text()

async def main(urls, concurrency=20):
    semaphore = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

Semaphore(20) garantisce che non ci siano mai più di 20 richieste attive contemporaneamente. È la tua leva principale di «cortesia»: calibra il valore in modo da non sovraccaricare il sito target. Aggiungi piccole pause casuali (await asyncio.sleep(random.uniform(0.1, 0.5))) per un ritmo più naturale.

6. Proxy

In aiohttp il proxy si passa come parametro della richiesta:

async with session.get(url, proxy="http://user:pass@ip:port") as resp:
    html = await resp.text()

La rotazione è semplicemente la scelta di un proxy casuale a ogni richiesta:

import random

PROXIES = ["http://ip1:port", "http://ip2:port", "http://ip3:port"]

async def fetch(session, url):
    proxy = random.choice(PROXIES)
    async with session.get(url, proxy=proxy) as resp:
        return await resp.text()

La strategia generale di lavoro con i proxy (tipi, scarto di quelli morti) è nella guida generale, sezione «Proxy».

7. Scraping attraverso TOR

aiohttp non supporta SOCKS in modo nativo: serve il pacchetto aiohttp-socks:

# pip install aiohttp-socks
import aiohttp
from aiohttp_socks import ProxyConnector

async def main(urls):
    connector = ProxyConnector.from_url("socks5://127.0.0.1:9050")
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as resp:
            print(await resp.json())   # IP del nodo di uscita di TOR

Il cambio del nodo di uscita tramite il segnale NEWNYM (libreria stem) è descritto nella guida generale, sezione «TOR». Tieni presente che TOR è lento: con una concorrenza alta diventerà il collo di bottiglia.

8. HTTPS/SSL

Di default aiohttp verifica i certificati. Puoi disattivare la verifica (solo per il debug) o fornire un contesto personalizzato:

import ssl

# disattivare la verifica — NON per la produzione
async with session.get(url, ssl=False) as resp:
    ...

# contesto SSL personalizzato
ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
async with session.get(url, ssl=ctx) as resp:
    ...

I principi di sicurezza della connessione sono nella guida generale, sezione «HTTPS/SSL».

9. Lavorare con i cookie

ClientSession conserva i cookie tra le richieste in automatico, come requests.Session:

async with aiohttp.ClientSession() as session:
    # login: il server imposta il cookie di sessione
    await session.post("https://example.com/login",
                       data={"user": "u", "pass": "p"})
    # le richieste successive sono già autenticate
    async with session.get("https://example.com/profile") as resp:
        html = await resp.text()

Puoi anche passare i cookie manualmente con il parametro cookies={...}. Maggiori dettagli nella guida generale, sezione «Cookie».

10. Stato della risposta e header

async with session.get(url) as resp:
    print(resp.status)                       # 200, 404 ...
    print(resp.headers.get("Content-Type"))
    if resp.status == 429:
        wait = int(resp.headers.get("Retry-After", 60))
        await asyncio.sleep(wait)            # non blocca gli altri task!
    resp.raise_for_status()

Il vantaggio chiave: nel gestire un 429, await asyncio.sleep() addormenta solo quella coroutine, mentre le altre continuano a lavorare. Nel codice sincrono time.sleep() congelerebbe tutto. La logica dei codici di stato è nella guida generale.

11. Code: asyncio.Queue

Per il crawling «man mano che si scoprono i link» si usano asyncio.Queue e un pool di worker-coroutine:

import asyncio
import aiohttp

async def worker(name, queue, session, visited):
    while True:
        url = await queue.get()
        if url not in visited:
            visited.add(url)
            try:
                async with session.get(url) as resp:
                    html = await resp.text()
                # ... trovare i nuovi link e metterli in coda:
                # for link in extract_links(html):
                #     await queue.put(link)
            except Exception as exc:
                print(f"{name} errore su {url}: {exc}")
        queue.task_done()

async def crawl(start_urls, num_workers=10):
    queue = asyncio.Queue()
    visited = set()
    for url in start_urls:
        queue.put_nowait(url)

    async with aiohttp.ClientSession() as session:
        workers = [asyncio.create_task(worker(f"w{i}", queue, session, visited))
                   for i in range(num_workers)]
        await queue.join()          # aspettiamo che la coda si svuoti
        for w in workers:
            w.cancel()

Il set serve per la deduplicazione, la Queue coordina i worker: è l’analogo asincrono del frontier della guida generale. Per una scansione distribuita la coda si sposta in Redis. L’implementazione industriale di questo schema la offre Scrapy (che al suo interno è anch’esso asincrono).

12. httpx come alternativa

httpx è un client moderno con la stessa API per codice sincrono e asincrono e con il supporto di HTTP/2:

import httpx
import asyncio

async def main(urls):
    async with httpx.AsyncClient(http2=True, timeout=15) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return responses

Se vuoi lo stesso codice commutabile tra sync e async e ti serve HTTP/2, httpx è più comodo di aiohttp. In velocità pura sui grandi volumi si equivalgono.

13. Pro e contro dello scraping asincrono

Pro:

• Concorrenza enorme con un consumo di memoria minimo.
• Accelerazione di svariate volte sulle attività I/O-bound.
• «Pause» a basso costo: asyncio.sleep non blocca gli altri task.
• Controllo fine della velocità tramite i semafori.

Contro:

• Più difficile da scrivere e da debuggare (async/await dappertutto).
• Il parsing CPU-bound blocca comunque il loop: serve un pool di processi.
• Non si può mescolare con librerie bloccanti senza run_in_executor.
• È facile sovraccaricare il sito target: serve disciplina con i semafori.

Quando sceglierlo: migliaia di pagine o più, e la velocità conta. Per un paio di centinaia di pagine è più semplice requests + ThreadPoolExecutor. Per un crawling completo di un intero sito c’è Scrapy, dove asincronia e code sono già integrate. E se lo scraper vive dentro un’applicazione web, tieni conto che l’ORM di Django resta in gran parte sincrono e lì il codice async richiede attenzione (sync_to_async): ne parliamo in «Web scraping con Django».