Raspado web asíncrono con AIOHTTP en Python

En esta guía, descubrirás:

• Qué es AIOHTTP y cuáles son las principales funciones que ofrece
• Una sección paso a paso sobre el uso de AIOHTTP para el raspado web
• Técnicas avanzadas para el raspado web con AIOHTTP
• Una comparación entre AIOHTTP y Requests para gestionar solicitudes automatizadas

¡Vamos allá!

¿Qué es AIOHTTP?

AIOHTTP es un marco HTTP asíncrono cliente/servidor creado sobre el asynciode Python. A diferencia de los clientes HTTP tradicionales, AIOHTTP utiliza sesiones de cliente para mantener las conexiones entre varias solicitudes. Por ello, es una opción eficaz para tareas basadas en sesiones de alta concurrencia.

⚙️Funciones

• Es compatible tanto con el lado cliente como con el lado servidor del protocolo HTTP.
• Ofrece compatibilidad nativa para WebSockets (tanto de cliente como de servidor).
• Ofrece middleware y enrutamiento conectable para servidores web.
• Gestiona eficazmente la transmisión de grandes cantidades de datos.
• Incluye la persistencia de la sesión del cliente, lo que permite reutilizar la conexión y reducir la sobrecarga en caso de varias solicitudes.

Raspado web con AIOHTTP: tutorial paso a paso

En el contexto del raspado web, AIOHTTP es solo un cliente HTTP para obtener los contenidos HTML sin procesar de una página. Para analizar y extraer datos de ese HTML, necesitas un analizador HTML como BeautifulSoup.

¡Sigue esta sección para aprender a usar AIOHTTP para realizar raspado web con BeautifulSoup!

Importante: Aunque AIOHTTP se usa principalmente en las etapas iniciales del proceso, te guiaremos a lo largo de todo el flujo de trabajo de raspado. Si te interesan las técnicas de raspado web AIOHTTP más avanzadas, no dudes en pasar al siguiente capítulo después del paso 3.

Paso 1: configura tu proyecto de raspado

Comprueba que Python 3+ está instalado en tu equipo. Si no lo está, descárgalo del sitio oficial y sigue las instrucciones de instalación.

A continuación, crea un directorio para tu proyecto de raspado AIOHTTP con este comando:

mkdir aiohttp-scraper

Navega hasta ese directorio y configura un  entorno virtual:

cd aiohttp-scraper
python -m venv env

Abre la carpeta del proyecto en tu IDE de Python preferido. Visual Studio Code con la extensión Python o PyCharm Community Edition son ambas opciones válidas.

A continuación, crea un archivo scraper.py dentro de la carpeta del proyecto. Al principio estará vacío, pero pronto le añadirás la lógica de raspado.

En la terminal de tu IDE, activa el entorno virtual. En Linux o macOS, usa:

./env/bin/activate

De manera equivalente, en Windows, ejecuta:

env/Scripts/activate

¡Genial! Ya está todo configurado y listo para comenzar.

Paso 2: configura las bibliotecas de raspado

Con el entorno virtual activado, instala AIOHTTP y BeautifulSoup con el siguiente comando:

pip install aiohttp beautifulsoup4

De este modo se añadirán tanto aiohttp como beautifulsoup4 a las dependencias de tu proyecto.

Impórtalos a tu script scraper.py: 

import asyncio
import aiohttp 
from bs4 import BeautifulSoup

Ten en cuenta que aiohttp requiere el asyncio para funcionar.

Ahora, añade el siguiente flujo de trabajo de funciones asíncronas a tu archivo scrper.py. 

async def scrape_quotes():
    # Scraping logic...

# Run the asynchronous function
asyncio.run(scrape_quotes())

scrape_quotes() define una función asíncrona en la que la lógica de raspado se ejecuta simultáneamente sin bloqueos. Por último, asyncio.run (scrape_quotes()) inicia y ejecuta la función asíncrona.

¡Genial! Puedes continuar con el siguiente paso de tu flujo de trabajo de raspado.

Paso 3: obtén el HTML de la página objetivo

En este ejemplo, aprenderás a raspar datos del sitio «Quotes to Scrape»:

Con bibliotecas como Requests o AIOHTTP, basta con hacer una solicitud GET para recibir directamente los contenidos HTML de la página. Sin embargo, AIOHTTP sigue un ciclo de vida de solicitud diferente.

El componente principal de AIOHTTP es ClientSession, que gestiona un conjunto de conexiones y admite Keep-Alive de forma predeterminada. En lugar de abrir una conexión nueva para cada solicitud, reutiliza las conexiones, lo que mejorará el rendimiento.

Al hacer una solicitud, el proceso normalmente consta de tres pasos:

1. Abrir una sesión a través de clientSession().
2. Enviar la solicitud GET de forma asíncrona con session.get().
3. Acceder a los datos de respuesta con métodos como await response.text().

Este diseño permite que el bucle de eventos utilice diferentes contextos entre operaciones sin bloquearse, lo que lo hace ideal para tareas de alta concurrencia.

Así, puedes usar AIOHTTP para recuperar el HTML de la página de inicio con esta lógica:

async with aiohttp.ClientSession() as session:
    async with session.get("http://quotes.toscrape.com") as response:
        # Access the HTML of the target page
        html = await response.text()

En segundo plano, AIOHTTP envía la solicitud al servidor y espera la respuesta, que contiene el HTML de la página. Una vez recibida la respuesta, await response.text() extrae los contenidos HTML en forma de cadena.

Imprime la variable html y verás:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Quotes to Scrape</title>
    <link rel="stylesheet" href="/static/bootstrap.min.css">
    <link rel="stylesheet" href="/static/main.css">
</head>
<body>
    <!-- omitted for brevity... -->
</body>
</html>

¡Así se hace! Has recuperado correctamente los contenidos HTML de la página objetivo. Es hora de analizar este contenido y extraer los datos que necesitas.

Paso 4: analiza el HTML

Pasa los contenidos HTML al constructor BeautifulSoup para analizarlos:

# Parse the HTML content using BeautifulSoup
soup = BeautifulSoup(html, "html.parser")

html.parser es el analizador HTML de Python predeterminado que se utiliza para procesar los contenidos.

El objeto soup contiene el HTML analizado y proporciona métodos para extraer los datos que necesitas.

AIOHTTP se ha ocupado de recuperar el HTML y ahora estás pasando a la fase típica de análisis de datos con BeautifulSoup. Para obtener más información, lee nuestro tutorial sobre BeautifulSoup web scraping.

Paso 5: escribe la lógica de extracción de datos

Puedes raspar los datos de las citas de la página usando el siguiente código:

# Where to store the scraped data
quotes = []

# Extract all quotes from the page
quote_elements = soup.find_all("div", class_="quote")

# Loop through quotes and extract text, author, and tags
for quote_element in quote_elements:
    text = quote_element.find("span", class_="text").get_text().get_text().replace("“", "").replace("”", "")
    author = quote_element.find("small", class_="author")
    tags = [tag.get_text() for tag in quote_element.find_all("a", class_="tag")]

    # Store the scraped data
    quotes.append({
        "text": text,
        "author": author,
        "tags": tags
    })

Este fragmento inicializa una lista denominada quotes para contener los datos raspados. A continuación, identifica todos los elementos HTML de la cita y los recorre para extraer el texto, el autor y las etiquetas de la cita. Cada cita extraída se almacena como un diccionario en la lista quotes, que organiza los datos para usarlos o exportarlos más adelante.

¡Estupendo! Ya está implementada la lógica de raspado.

Paso 6: exporta los datos raspados

Usa estas líneas de código para exportar los datos raspados a un archivo CSV:

# Open the file for export
with open("quotes.csv", mode="w", newline="", encoding="utf-8") as file:
    writer = csv.DictWriter(file, fieldnames=["text", "author", "tags"])
    
    # Write the header row
    writer.writeheader()
    
    # Write the scraped quotes data
    writer.writerows(quotes)

El fragmento anterior abre un archivo llamado quotes.csv en modo de escritura. A continuación, configura los encabezados de las columnas (text, author y tags), escribe los encabezados y, a continuación, escribe cada diccionario de la lista quotes en el archivo CSV.

csv.DictWriter simplifica el formato de los datos, lo que, a su vez, facilita el almacenamiento de datos estructurados. Para que funcione, recuerda importar csv de la biblioteca estándar de Python:

import csv

Paso 7: júntalo todo

Este es el aspecto que debería tener tu script final de raspado web AIOHTTP:

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import csv

# Define an asynchronous function to make the HTTP GET request
async def scrape_quotes():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://quotes.toscrape.com") as response:
            # Access the HTML of the target page
            html = await response.text()

            # Parse the HTML content using BeautifulSoup
            soup = BeautifulSoup(html, "html.parser")

            # List to store the scraped data
            quotes = []

            # Extract all quotes from the page
            quote_elements = soup.find_all("div", class_="quote")

            # Loop through quotes and extract text, author, and tags
            for quote_element in quote_elements:
                text = quote_element.find("span", class_="text").get_text().replace("“", "").replace("”", "")
                author = quote_element.find("small", class_="author").get_text()
                tags = [tag.get_text() for tag in quote_element.find_all("a", class_="tag")]

                # Store the scraped data
                quotes.append({
                    "text": text,
                    "author": author,
                    "tags": tags
                })

            # Open the file name for export
            with open("quotes.csv", mode="w", newline="", encoding="utf-8") as file:
                writer = csv.DictWriter(file, fieldnames=["text", "author", "tags"])

                # Write the header row
                writer.writeheader()

                # Write the scraped quotes data
                writer.writerows(quotes)

# Run the asynchronous function
asyncio.run(scrape_quotes())

Puedes ejecutarlo con:

python scraper.py

O, en Linux/macOS:

python3 scraper.py

Aparecerá un archivo quotes.csv en la carpeta raíz de tu proyecto. Ábrelo y verás:

¡Ya está! Acabas de aprender a realizar el raspado web con AIOHTTP y BeautifulSoup.

AIOHTTP para el raspado web: características y técnicas avanzadas

Ahora que sabes usar AIOHTTP para el raspado web básico, es hora de ver casos más avanzados.

En los siguientes ejemplos, el sitio objetivo será el punto final HTTPBin.io /anything.  Es una API práctica que devuelve la dirección IP, los encabezados y otros datos que envíe el solicitante.

¡Prepárate para dominar AIOHTTP para realizar raspado web!

Configura encabezados personalizados

Puedes especificar encabezados personalizados en una solicitud AIOHTTP con el argumento headers: 

import aiohttp
import asyncio

async def fetch_with_custom_headers():
    # Custom headers for the request
    headers = {
        "Accept": "application/json",
        "Accept-Language": "en-US,en;q=0.9,fr-FR;q=0.8,fr;q=0.7,es-US;q=0.6,es;q=0.5,it-IT;q=0.4,it;q=0.3"
    }

    async with aiohttp.ClientSession() as session:
        # Make a GET request with custom headers
        async with session.get("https://httpbin.io/anything", headers=headers) as response:
            data = await response.json()
            # Handle the response...
            print(data)

# Run the event loop
asyncio.run(fetch_with_custom_headers())

De esta forma, AIOHTTP realizará una solicitud GET HTTP con los encabezados Accept y Accept-Language configurados.

Establece un User-Agent personalizado

User-Agent es uno de los encabezados HTTP más importantes para realizar raspado web. Por defecto, AIOHTTP usa este User-Agent:

Python/<PYTHON_VERSION> aiohttp/<AIOHTTP_VERSION>

El valor predeterminado anterior puede mostrar fácilmente que tus solicitudes proceden de un script automatizado. Eso aumenta el riesgo de que el sitio objetivo te bloquee.

Para reducir las probabilidades de que te detecten, puedes configurar un User-Agent personalizado para el mundo real como antes:

import aiohttp
import asyncio

async def fetch_with_custom_user_agent():
    # Define a Chrome-like custom User-Agent
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"
    }

    async with aiohttp.ClientSession(headers=headers) as session:
        # Make a GET request with the custom User-Agent
        async with session.get("https://httpbin.io/anything") as response:
            data = await response.text()
            # Handle the response...
            print(data)

# Run the event loop
asyncio.run(fetch_with_custom_user_agent())

¡Descubre los mejores agentes de usuario para el raspado web!

Configura cookies

Al igual que los encabezados HTTP, puedes configurar cookies personalizadas usando las cookies en ClientSession():

import aiohttp
import asyncio

async def fetch_with_custom_cookies():
    # Define cookies as a dictionary
    cookies = {
        "session_id": "9412d7hdsa16hbda4347dagb",
        "user_preferences": "dark_mode=false"
    }

    async with aiohttp.ClientSession(cookies=cookies) as session:
        # Make a GET request with custom cookies
        async with session.get("https://httpbin.io/anything") as response:
            data = await response.text()
            # Handle the response...
            print(data)

# Run the event loop
asyncio.run(fetch_with_custom_cookies())

Las cookies te ayudan a incluir los datos de sesión necesarios en tus solicitudes de raspado web.

Ten en cuenta que las cookies configuradas en ClientSession se comparten en todas las solicitudes realizadas en esa sesión. Para acceder a las cookies de sesión, consulta ClientSession.cookie_jar.

Integración de proxy

En AIOHTTP, puedes enrutar tus solicitudes a través de un servidor proxy para reducir el riesgo de prohibiciones de IP. Para ello, utiliza el argumento proxy  en la función del método HTTP en la sesión:

import aiohttp
import asyncio

async def fetch_through_proxy():
    # Replace with the URL of your proxy server
    proxy_url = "<YOUR_PROXY_URL>"

    async with aiohttp.ClientSession() as session:
        # Make a GET request through the proxy server
        async with session.get("https://httpbin.io/anything", proxy=proxy_url) as response:
            data = await response.text()
            # Handle the response...
            print(data)

# Run the event loop
asyncio.run(fetch_through_proxy())

Descubre cómo realizar la autenticación y la rotación de un proxy con nuestra guía sobre cómo usar un proxy en AIOHTTP.

Gestión de errores

Por defecto, AIOHTTP solo genera errores por problemas de conexión o de red. Para generar excepciones para las respuestas HTTP al recibir los códigos de estado 4xx y 5xx, puedes usar cualquiera de los siguientes métodos:

1. Establecer raise_for_status=True al crear la ClientSession: genera excepciones automáticamente para todas las solicitudes realizadas durante la sesión si el estado de la respuesta es 4xx o 5xx.
2. Pasar raise_for_status=True directamente a los métodos de solicitud: habilita la generación de errores para cada método de solicitud (como session.get() o session.post()) sin afectar a los demás.
3. Llamar manualmente a response.raise_for_status(): controla completamente cuándo generar excepciones, lo que te permitirá decidir por solicitud.

Ejemplo de opción 1:

import aiohttp
import asyncio

async def fetch_with_session_error_handling():
    async with aiohttp.ClientSession(raise_for_status=True) as session:
        try:
            async with session.get("https://httpbin.io/anything") as response:
                # No need to call response.raise_for_status(), as it is automatic
                data = await response.text()
                print(data)
        except aiohttp.ClientResponseError as e:
            print(f"HTTP error occurred: {e.status} - {e.message}")
        except aiohttp.ClientError as e:
            print(f"Request error occurred: {e}")

# Run the event loop
asyncio.run(fetch_with_session_error_handling())

Cuando raise_for_status=True se establece en el nivel de sesión, todas las solicitudes realizadas a través de esa sesión generarán un aiohttp.ClientResponseError en las respuestas 4xx o 5xx. 

Ejemplo de opción 2:

import aiohttp
import asyncio

async def fetch_with_raise_for_status():
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get("https://httpbin.io/anything", raise_for_status=True) as response:
                # No need to manually call response.raise_for_status(), it is automatic
                data = await response.text()
                print(data)
        except aiohttp.ClientResponseError as e:
            print(f"HTTP error occurred: {e.status} - {e.message}")
        except aiohttp.ClientError as e:
            print(f"Request error occurred: {e}")

# Run the event loop
asyncio.run(fetch_with_raise_for_status())

En este caso, el argumento raise_for_status=True se pasa directamente a la llamada session.get().  Así se garantiza que se genere automáticamente una excepción para cualquier código de estado 4xx o 5xx. 

Ejemplo de opción 3:

import aiohttp
import asyncio

async def fetch_with_manual_error_handling():
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get("https://httpbin.io/anything") as response:
                response.raise_for_status()  # Manually raises error for 4xx/5xx
                data = await response.text()
                print(data)
        except aiohttp.ClientResponseError as e:
            print(f"HTTP error occurred: {e.status} - {e.message}")
        except aiohttp.ClientError as e:
            print(f"Request error occurred: {e}")

# Run the event loop
asyncio.run(fetch_with_manual_error_handling())

Si prefieres tener más control sobre cada solicitud, puedes llamar a response.raise_for_status() manualmente tras hacer una solicitud. Este enfoque te permite decidir exactamente cuándo gestionar los errores.

Reintenta las solicitudes fallidas

AIOHTTP no proporciona compatibilidad integrada para reintentar solicitudes automáticamente. Para implementarlo, debes usar una lógica personalizada o una biblioteca de terceros como aiohttp-retry. De este modo, podrás configurar la lógica de reintento para las solicitudes fallidas, lo que te ayudará a gestionar los problemas transitorios de la red, los tiempos de espera o los límites de velocidad.

Instala aiohttp-retry con:

pip install aiohttp-retry

Luego, puedes usarlo de la siguiente manera:

import asyncio
from aiohttp_retry import RetryClient, ExponentialRetry

async def main():
    retry_options = ExponentialRetry(attempts=1)
    retry_client = RetryClient(raise_for_status=False, retry_options=retry_options)
    async with retry_client.get("https://httpbin.io/anything") as response:
        print(response.status)
        
    await retry_client.close()

Así se configura el comportamiento de reintento, con una estrategia de retroceso exponencial. Obtén más información en los documentos oficiales.

Comparativa de AIOHTTP y Requests para realizar raspado web

A continuación, se muestra una tabla resumida para comparar AIOHTTP y Requests para realizar raspado web:

Para ver una comparación completa, consulta nuestra entrada de blog sobre Comparativa de Requests, HTTPX y AIOHTTP.

Aprende a raspar sitios web con HTTPX.

Conclusión

En este artículo, has aprendido a utilizar la biblioteca aiohttp para realizar el raspado web. Has analizado lo que es, las funciones que ofrece y las ventajas que aporta. AIOHTTP destaca por ser una opción rápida y fiable para realizar solicitudes HTTP al recopilar datos en línea.

Sin embargo, las solicitudes HTTP automatizadas exponen tu dirección IP pública. De este modo, tu identidad y tu ubicación pueden quedar al descubierto, poniendo en peligro tu privacidad. Para proteger tu seguridad y privacidad, una de las estrategias más eficaces es utilizar un servidor proxy para ocultar tu dirección IP.

Bright Data controla los mejores servidores proxy del mundo y trabaja para empresas de la lista Fortune 500 y para más de 20 000 clientes. Su oferta incluye una gran variedad de tipos de proxy:

• Proxies de centros de datos: más de 770 000 IP de centros de datos.
• Proxies residenciales: más de 72 millones de IP residenciales en más de 195 países.
• Proxies de ISP: más de 700 000 IP de ISP.
• Proxies móviles: más de 7 millones de IP móviles.

¡Crea una cuenta gratuita de Bright Data hoy mismo para probar nuestros proxies y soluciones de raspado web!