Scraping Web asynchrone avec AIOHTTP en Python

Dans ce guide, vous allez découvrir :

• Ce qu’est AIOHTTP et les principales fonctionnalités qu’il offre
• Une section étape par étape sur l’utilisation d’AIOHTTP pour le web scraping
• Techniques avancées pour le web scraping avec AIOHTTP
• Une comparaison entre AIOHTTP et Requests pour gérer les requêtes automatisées

C’est parti !

Qu’est-ce qu’AIOHTTP ?

AIOHTTP est un framework HTTP client/serveur asynchrone construit sur la base d’ asyncio de Python. Contrairement aux clients HTTP traditionnels, AIOHTTP utilise des sessions client pour maintenir les connexions entre plusieurs requêtes. Cela en fait un choix efficace pour les tâches basées sur des sessions à haute concordance.

⚙️ Caractéristiques

• Prend en charge les parties client et serveur du protocole HTTP.
• Fournit un support natif pour les WebSockets (à la fois pour le client et le serveur).
• Fournit un intergiciel et un routage enfichable pour les serveurs web.
• Traite efficacement les données volumineuses en continu.
• Inclut la persistance de la session client, permettant la réutilisation de la connexion et réduisant la charge de travail pour les requêtes multiples.

Scraping avec AIOHTTP : tutoriel étape par étape

Dans le contexte du web scraping, AIOHTTP est simplement un client HTTP permettant de récupérer le contenu HTML brut d’une page. Pour analyser et extraire les données de ce HTML, vous avez besoin d’un analyseur HTML comme BeautifulSoup.

Suivez cette section pour apprendre à utiliser AIOHTTP pour le web scraping avec BeautifulSoup !

Avertissement: Bien que AIOHTTP soit utilisé principalement dans les premières étapes du processus, nous vous guiderons tout au long du processus de scraping. Si vous êtes intéressé par des techniques plus avancées de web scraping AIOHTTP, n’hésitez pas à passer au chapitre suivant après l’étape 3.

Étape n° 1 : configurer votre projet de scraping

Assurez-vous que Python 3+ est installé sur votre machine. Si ce n’est pas le cas, téléchargez-le sur le site officiel et suivez les instructions d’installation.

Ensuite, créez un répertoire pour votre projet de scraping AIOHTTP en utilisant cette commande :

mkdir aiohttp-scraper

Naviguez dans ce répertoire et mettez en place un environnement virtuel:

cd aiohttp-scraper
python -m venv env

Ouvrez le dossier du projet dans votre IDE Python préféré. Visual Studio Code avec l’extension Python et PyCharm Community Edition sont deux choix valables.

Créez maintenant un fichier scraper.py dans le dossier du projet. Il sera vide au début, mais vous y ajouterez bientôt la logique de scraping.

Sur le terminal de votre IDE, activez l’environnement virtuel. Sous Linux ou macOS, utilisez :

./env/bin/activate

De manière équivalente, sous Windows, exécutez :

env/Scripts/activate

Parfait ! Nous allons pouvoir continuer.

Étape n° 2 : configurer les bibliothèques de scraping

Avec l’environnement virtuel activé, installez AIOHTTP et BeautifulSoup en utilisant la commande ci-dessous :

pip install aiohttp beautifulsoup4

Ceci ajoutera les bibliothèques aiohttp et beautifulsoup4 aux dépendances de votre projet.

Importez-les dans votre script scraper.py :

import asyncio
import aiohttp 
from bs4 import BeautifulSoup

Notez que aiohttp nécessite asyncio pour fonctionner.

Ajoutez maintenant la fonction async suivante à votre fichier scrper.py :

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

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

scrape_quotes() définit une fonction asynchrone dans laquelle votre logique de scraping s’exécutera simultanément sans blocage. Enfin, asyncio.run(scrape_quotes()) démarre et exécute la fonction asynchrone.

Excellent ! Vous pouvez passer à l’étape suivante de votre flux de travail.

Étape 3 : Obtenir le code HTML de la page cible

Dans cet exemple, nous allons voir comment récupérer les données du site « Quotes to Scrape » :

Avec des bibliothèques comme Requests ou AIOHTTP, il suffit de faire une requête GET et de recevoir directement le contenu HTML de la page. Cependant, AIOHTTP suit un cycle de vie des demandes différent.

Le composant principal d’AIOHTTP est le ClientSession, qui gère un pool de connexions et prend en charge la fonction Keep-Alive par défaut. Au lieu d’ouvrir une nouvelle connexion pour chaque demande, il réutilise les connexions, ce qui améliore les performances.

Lorsqu’une demande est formulée, le processus comporte généralement trois étapes :

1. Ouverture d’une session par ClientSession().
2. Envoi de la requête GET de manière asynchrone avec session.get().
3. Accès aux données de la réponse avec des méthodes telles que await response.text().

Cette conception permet à la boucle d’événements d’utiliser différents contextes entre les opérations sans blocage, ce qui la rend idéale pour les tâches répétées fréquemment.

Vous pouvez ainsi utiliser AIOHTTP pour récupérer le code HTML de la page d’accueil avec cette logique :

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 coulisses, AIOHTTP envoie la requête au serveur et attend la réponse, qui contient le code HTML de la page. Une fois la réponse reçue, await response.text() extrait le contenu HTML sous forme de chaîne.

Imprimez la variable html et vous verrez :

<!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>

Super ! Vous avez réussi à récupérer le contenu HTML de la page cible. Il est temps d’analyser ce contenu et d’en extraire les données dont vous avez besoin.

Étape 4 : Analyse du code HTML

Passer le contenu HTML dans le constructeur BeautifulSoup pour l’analyser :

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

html.parser est l’analyseur HTML Python par défaut utilisé pour traiter le contenu.

L’objet soup contient le code HTML parsé et fournit des méthodes pour extraire les données dont vous avez besoin.

AIOHTTP s’est occupé de récupérer le HTML. Nous allons maintenant passer à la phase typique de parsing des données avec BeautifulSoup. Pour en savoir plus, suivez notre tutoriel sur le web scraping avec BeautifulSoup.

Etape #5 : Ecrire la logique d’extraction des données

Vous pouvez scraper les données de la page en utilisant le code suivant :

# 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
    })

Ce snippet initialise une liste nommée quotes qui contiendra les données scrapées. Tous les éléments HTML de la citation sont ensuite analysés et parcourus en boucle pour en extraire le texte de la citation, l’auteur et les balises. Chaque citation extraite est stockée sous forme de dictionnaire dans la liste des citations , ce qui permet d’organiser les données en vue de les utiliser et de les exporter ultérieurement.

Super ! La logique de scraping est maintenant implémentée.

Étape 6 : exporter les données scrapées

Utilisez ces lignes de code pour exporter les données récupérées dans un fichier 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)

L’extrait ci-dessus ouvre un fichier nommé quotes.csv en mode écriture. Ensuite, il configure les en-têtes de colonnes (texte, auteur, tags), écrit les en-têtes, puis écrit chaque dictionnaire de la liste des citations dans le fichier CSV.

csv.DictWriter simplifie le formatage des données, ce qui facilite le stockage de données structurées. Pour que cela fonctionne, n’oubliez pas d’importer csv de la bibliothèque standard de Python :

import csv

Étape n° 7 : assembler le tout

Voici à quoi devrait ressembler votre script final de scraping 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())

Vous pouvez l’exécuter avec :

python scraper.py

Ou, sous Linux/macOS :

python3 scraper.py

Un fichier quotes.csv apparaît dans le dossier racine de votre projet. Ouvrez le fichier et vous verrez :

Et voilà ! Vous venez d’apprendre à faire du web scraping avec AIOHTTP et BeautifulSoup.

AIOHTTP pour le Web Scraping : Fonctionnalités et techniques avancées

Maintenant que vous savez comment utiliser AIOHTTP pour le web scraping de base, il est temps de voir des scénarios plus avancés.

Dans les exemples suivants, le site cible sera le point de terminaison HTTPBin.io /anything. Il s’agit d’une API pratique qui renvoie l’adresse IP, les en-têtes et d’autres données envoyées par le demandeur.

Préparez-vous à maîtriser AIOHTTP pour le web scraping !

Définit des en-têtes personnalisés

Vous pouvez spécifier des en-têtes personnalisés dans une requête AIOHTTP avec l’argument en-têtes :

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 cette manière, AIOHTTP effectuera une requête HTTP GET avec les paramètres Accept et Accept-Language .

Définit un agent utilisateur personnalisé

User-Agent est l’un des en-têtes HTTP les plus critiques pour le web scraping. Par défaut, AIOHTTP utilise ce User-Agent:

Python/<PYTHON_VERSION> aiohttp/<AIOHTTP_VERSION>

La valeur par défaut ci-dessus peut facilement faire apparaître vos demandes comme provenant d’un script automatisé. Cela augmentera le risque d’être bloqué par le site cible.

Pour réduire les risques d’être détecté, vous pouvez définir un véritable User-Agent personnalisé comme précédemment :

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())

Découvrez les meilleurs agents utilisateurs pour le web scraping!

Définir les cookies

Tout comme les en-têtes HTTP, vous pouvez définir des cookies personnalisés en utilisant les cookies dans 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())

Les cookies vous permettent d’inclure les données de session requises dans vos requêtes de web scraping.

Notez que les cookies définis dans ClientSession sont partagés par toutes les requêtes effectuées avec cette session. Pour accéder aux cookies de session, consultez ClientSession.cookie_jar.

Intégration de proxies

Dans AIOHTTP, vous pouvez faire passer vos requêtes par un serveur proxy afin de réduire le risque d’interdiction d’IP. Pour ce faire, utilisez l’argument proxy  dans la fonction HTTP method sur session:

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())

Découvrez comment effectuer l’authentification et la rotation du proxy dans notre guide sur comment utiliser un proxy dans AIOHTTP.

Gestion des erreurs

Par défaut, AIOHTTP ne génère des erreurs qu’en cas de problèmes de connexion ou de réseau. Pour lever des exceptions pour les réponses HTTP lors de la réception de codes d’état 4xx et 5xx, vous pouvez utiliser l’une des approches suivantes :

1. Définir raise_for_status=True lors de la création du ClientSession : cela lève automatiquement des exceptions pour toutes les demandes effectuées par l’intermédiaire de la session si le statut de la réponse est 4xx ou 5xx.
2. Transmettre raise_for_status=True directement aux méthodes de requête : cela active la levée d’erreurs pour des méthodes de requête individuelles (comme session.get() ou session.post()) sans affecter les autres.
3. Appeler response.raise_for_status() manuellement : permet de contrôler entièrement le moment où les exceptions sont levées, en vous permettant de décider au cas par cas.

Exemple de l’option 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())

Lorsque raise_for_status=True est défini au niveau de la session, toutes les requêtes effectuées via cette session lèveront une erreur aiohttp.ClientResponseError pour les réponses 4xx or 5xx.

Exemple de l’option 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())

Dans ce cas, l’argument raise_for_status=True est transmis directement à l’appel session.get() . Cette option garantit qu’une exception est levée automatiquement pour tout code d’état 4xx ou 5xx.

Exemple de l’option 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 vous préférez avoir plus de contrôle sur les requêtes individuelles, vous pouvez appeler response.raise_for_status() manuellement après avoir effectué une requête. Cette approche vous permet de décider exactement quand traiter les erreurs.

Nouvelle tentative pour les requêtes qui ont échoué

AIOHTTP ne fournit pas de support intégré pour relancer automatiquement les requêtes. Pour ce faire, vous devez utiliser une logique personnalisée ou une bibliothèque tierce comme aiohttp-retry. Cela vous permet de configurer une logique de relance pour les demandes qui ont échoué, afin de gérer les problèmes de réseau transitoires, les dépassements de délai ou les limites de débit.

Installer aiohttp-retry avec :

pip install aiohttp-retry

Ensuite, vous pouvez l’utiliser comme suit :

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()

Ceci configure le comportement des tentatives, avec une stratégie de backoff exponentielle. Pour en savoir plus, consultez la documentation officielle.

AIOHTTP vs Requests pour le Web Scraping

Vous trouverez ci-dessous un tableau récapitulatif comparant AIOHTTP et Requests pour le web scraping :

Pour une comparaison complète, consultez notre article de blog sur Requêtes vs HTTPX vs AIOHTTP.

Apprenez à scraper des sites web avec HTTPX.

Conclusion

Dans cet article, nous avons appris à utiliser la bibliothèque aiohttp pour le web scraping. Vous avez exploré ce qu’il est, les caractéristiques qu’il offre et les avantages qu’il procure. AIOHTTP est un choix rapide et fiable pour effectuer des requêtes HTTP lors de la collecte de données en ligne.

Cependant, les requêtes HTTP automatisées révèlent votre adresse IP publique. Cela peut révéler votre identité et votre localisation, mettant ainsi votre vie privée en danger. Pour préserver votre sécurité et votre vie privée, l’une des stratégies les plus efficaces consiste à utiliser un serveur proxy pour cacher votre adresse IP.

Bright Data contrôle les meilleurs serveurs proxy au monde, au service des entreprises du Fortune 500 et de plus de 20 000 clients. Son offre comprend une large gamme de types de proxy :

• Proxys de centre de données — Plus de 770 000 adresses IP de datacenters.
• Proxys résidentiels – Plus de 72 millions d’adresses IP résidentielles dans plus de 195 pays.
• Proxys de fournisseurs d’accès à Internet — Plus de 700 000 adresses IP de fournisseurs d’accès Internet.
• Proxies mobiles : plus de 7 millions d’adresses IP mobiles.

Créez un compte Bright Data gratuit dès aujourd’hui pour tester nos proxys et nos solutions de scraping !