Scraping Web Asíncrono con Python y Aiohttp

TL;DR

asyncio mejora el rendimiento para muchas esperas de red independientes; no hace más rápido el análisis de HTML ni otorga permiso para aumentar la tasa de solicitudes de un sitio.
Reutiliza una aiohttp.ClientSession, establece tiempos de espera totales y de conexión, y limita tanto la concurrencia global como por host.
Un semáforo limita las tareas simultáneas, no las solicitudes por segundo. Añade un ritmo explícito por host cuando la fuente lo requiera.
Vuelve a intentar solo fallos transitorios como 429, 502, 503, y 504, con un retraso limitado y manejo de Retry-After; no vuelvas a intentar ciegamente cada respuesta 4xx.
Para sitios renderizados por JavaScript o descubrimiento recursivo, llama a una API de renderizado y Crawl gestionada en lugar de esperar que aiohttp ejecute JavaScript.

¿Cuándo Deberías Usar Asyncio para Web Scraping?

Usa asyncio cuando el scraper pasa la mayor parte de su tiempo esperando muchas respuestas HTTP permitidas. Un trabajador puede gestionar múltiples solicitudes en vuelo sin un hilo por solicitud. El beneficio depende de la latencia de la fuente, la tasa de solicitud permitida, el costo de análisis y el almacenamiento posterior. Si la fuente permite solo una solicitud cada pocos segundos, un cliente asíncrono no puede legítimamente convertirlo en un rastreo de alta tasa. El producto Nstdata Crawl es una opción gestionada para los pasos de adquisición y conversión discutidos aquí.

import asyncio
import time
from collections import defaultdict
from urllib.parse import urlparse

import aiohttp
from bs4 import BeautifulSoup

URLS = ["https://example.com/"]  # Reemplaza con URLs aprobadas.
MAX_CONCURRENT = 8
MAX_PER_HOST = 3
MIN_HOST_INTERVAL = 1.0
RETRYABLE = {429, 502, 503, 504}


class HostPacer:
    def __init__(self, interval: float):
        self.interval = interval
        self.locks = defaultdict(asyncio.Lock)
        self.next_start = defaultdict(float)

    async def wait(self, host: str):
        async with self.locks[host]:
            delay = self.next_start[host] - time.monotonic()
            if delay > 0:
                await asyncio.sleep(delay)
            self.next_start[host] = time.monotonic() + self.interval


def retry_delay(header, attempt: int) -> float:
    if header and header.isdigit():
        return min(float(header), 60.0)
    return min(2 ** attempt, 30.0)


async def fetch(session, pacer, semaphore, url):
    host = urlparse(url).hostname
si not host o urlparse(url).scheme not in {"http", "https"}:
        raise ValueError(f"URL no soportada: {url}")

    for attempt in range(3):
        await pacer.wait(host)
        try:
            async with semaphore:
                async with session.get(url) as response:
                    if response.status in RETRYABLE:
                        delay = retry_delay(response.headers.get("Retry-After"), attempt)
                    else:
                        response.raise_for_status()
                        html = await response.text()
                        soup = BeautifulSoup(html, "html.parser")
                        title = soup.title.get_text(" ", strip=True) if soup.title else None
                        if not title:
                            raise ValueError(f"Título faltante: {url}")
                        return {"url": str(response.url), "status": response.status, "title": title}
        except (aiohttp.ClientError, asyncio.TimeoutError):
            if attempt == 2:
                raise
            delay = min(2 ** attempt, 30.0)

        if attempt == 2:
            raise RuntimeError(f"Límite de reintentos alcanzado: {url}")
        await asyncio.sleep(delay)


async def main():
    timeout = aiohttp.ClientTimeout(total=30, connect=10)
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT, limit_per_host=MAX_PER_HOST)
    headers = {"User-Agent": "ExampleResearchBot/1.0 (contacto: team@example.org)"}
    pacer = HostPacer(MIN_HOST_INTERVAL)
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    async with aiohttp.ClientSession(timeout=timeout, connector=connector, headers=headers) as session:
        results = await asyncio.gather(
            *(fetch(session, pacer, semaphore, url) for url in URLS),
            return_exceptions=True,
        )
    for url, result in zip(URLS, results):
        print(url, result)


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

Asyncio para Web Scraping: Peticiones Concurrentes Hechas Correctamente

TL;DR

¿Cuándo Deberías Usar Asyncio para Web Scraping?

Articulos mas populares

Asyncio para Web Scraping: Peticiones Concurrentes Hechas Correctamente

Web Scraping con Python: La Guía Completa de 2026

¿Qué Necesitas Antes de Comenzar?

¿Cómo Difere la Concurrencia del Limitado de Tasa?

¿Cómo Recuperas Páginas Concurrentemente con Aiohttp?

Método 1: Ejecutar un Recuperador de HTML Limitado

¿Cómo procesas miles de URLs sin crear miles de tareas a la vez?

Método 2: Usar una cola de trabajo limitada

¿Cuándo deberías usar Nstdata Crawl en lugar de Aiohttp?

Método 3: Delegar renderizado y descubrimiento

¿Por qué fallan o devuelven datos incorrectos las solicitudes concurrentes?

¿Cómo haces seguro para producción el ejemplo?

¿Qué deberías monitorear en un scraper concurrente?

Preguntas Frecuentes