Captura assíncrona da Web com AIOHTTP em Python

Neste guia, você explorará:

• O que é AIOHTTP e os principais recursos que ele fornece
• Uma seção passo a passo sobre como usar AIOHTTP para web scraping
• Técnicas avançadas para web scraping com AIOHTTP
• Uma comparação entre AIOHTTP e Requests para lidar com requisições automatizadas

Vamos lá!

O que é AIOHTTP?

AIOHTTP é um framework HTTP cliente/servidor assíncrono construído sobre o  asyncio do Python. Diferentemente dos clientes HTTP tradicionais, o AIOHTTP usa sessões de cliente para manter conexões em várias solicitações. Isso a torna uma opção eficiente para tarefas baseadas em sessões de alta concorrência.

⚙️ Recursos

• Suporta tanto o lado cliente quanto o lado servidor do protocolo HTTP.
• Fornece suporte nativo para WebSockets (cliente e servidor).
• Oferece middleware e roteamento plugável para servidores web.
• Lida com eficiência com o streaming de grandes volumes de dados.
• Inclui a persistência da sessão do cliente, permitindo a reutilização da conexão e reduzindo a sobrecarga de várias solicitações.

Scraping com AIOHTTP: Tutorial Passo a Passo

No contexto da web scraping, o AIOHTTP é apenas um cliente HTTP para buscar o conteúdo HTML bruto de uma página. Para analisar e extrair dados desse HTML, você precisará de um analisador de HTML como BeautifulSoup.

Siga esta seção para aprender como usar o AIOHTTP para web scraping com o BeautifulSoup!

Aviso: Embora o AIOHTTP seja usado principalmente nos estágios iniciais do processo, guiaremos você por todo o fluxo de trabalho de raspagem. Se você estiver interessado em técnicas mais avançadas de raspagem web AIOHTTP, sinta-se à vontade para pular para o próximo capítulo após a Etapa 3.

Etapa #1: Configure seu projeto de raspagem

Certifique-se de que o Python 3+ esteja instalado em sua máquina. Caso contrário, baixe-o do site oficial e siga as instruções de instalação.

Em seguida, crie um diretório para seu projeto de raspagem AIOHTTP usando este comando:

mkdir aiohttp-scraper

Navegue até esse diretório e configure um ambiente virtual:

cd aiohttp-scraper
python -m venv env

Abra a pasta do projeto no seu IDE Python preferido. Visual Studio Code com a extensão para Python ou PyCharm Community Edition são ambas escolhas válidas.

Agora, crie um arquivo scraper.py dentro da pasta do projeto. Ela ficará vazia no início, mas em breve você adicionará a lógica de raspagem a ela.

No terminal do seu IDE, ative o ambiente virtual. No Linux ou macOS, use:

./env/bin/activate

Equivalentemente, no Windows, execute:

env/Scripts/activate

Ótimo! Você já está completamente configurado e pronto para começar.

Passo #2: Configure as Bibliotecas de Scraping

Com o ambiente virtual ativado, instale o AIOHTTP e o BeautifulSoup usando o comando abaixo:

pip install aiohttp beautifulsoup4

Isso adicionará tanto o aiohttp quanto o beautifulsoup4 às dependências do seu projeto.

Importe-os para o seu scraper.py script:

import asyncio
import aiohttp 
from bs4 import BeautifulSoup

Note que aiohttp requer que o asyncio funcione.

Agora, adicione o seguinte fluxo de trabalho da função async ao seu arquivo scrper.py :

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

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

scrape_quotes() define uma função assíncrona em que sua lógica de raspagem será executada simultaneamente sem bloqueio. Finalmente, asyncio.run(scrape_quotes()) inicia e executa a função assíncrona.

Formidável! Você pode prosseguir para a próxima etapa em seu fluxo de trabalho de raspagem.

Etapa #3: Obtenha o HTML da página de destino

Neste exemplo, você verá como extrair dados do site “Quotes to Scrape“:

Com bibliotecas como Requests ou AIOHTTP, você simplesmente faria uma solicitação GET e receberia diretamente o conteúdo HTML da página. No entanto, o AIOHTTP segue um ciclo de vida de solicitação diferente.

O componente principal do AIOHTTP é o ClientSession, que gerencia um pool de conexões e suporta Keep-Alive por padrão. Em vez de abrir uma nova conexão para cada solicitação, ele reutiliza as conexões, melhorando o desempenho.

Ao fazer uma solicitação, o processo normalmente envolve três etapas:

1. Abrindo uma sessão por meio de ClientSession().
2. Enviando a solicitação GET de forma assíncrona com session.get().
3. Acessando os dados de resposta com métodos como await response.text().

Esse design permite que o loop de eventos use diferentes com contextos entre operações sem bloqueio, tornando-o ideal para tarefas de alta concorrência.

Diante disso, você pode usar AIOHTTP para recuperar o HTML da página inicial com 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()

Nos bastidores, o AIOHTTP envia a solicitação ao servidor e aguarda a resposta, que contém o HTML da página. Depois que a resposta for recebida, await response.text() extrai o conteúdo HTML como uma string.

Imprima a html variável e você verá:

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

Muito bem! Você recuperou com sucesso o conteúdo HTML da página de destino. É hora de analisar esse conteúdo e extrair os dados que você precisa.

Etapa #4: Analise o HTML

Passe o conteúdo HTML para o construtor BeautifulSoup para analisá-lo:

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

html.parser é o analisador HTML padrão do Python usado para processar o conteúdo.

O objeto soup contém o HTML analisado e fornece métodos para extrair os dados de que você precisa.

AIOHTTP lidou com a recuperação do HTML e agora você está fazendo a transição para a fase típica de análise de dados com o BeautifulSoup. Para mais detalhes, leia nosso tutorial sobre web scraping com BeautifulSoup.

Etapa #5: Escreva a lógica de extração de dados

Você pode extrair os dados das cotações da página usando o seguinte 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
    })

Esse trecho inicializa uma lista chamada quotes para conter os dados coletados. Em seguida, ele identifica todos os elementos HTML da citação e os percorre para extrair o texto, o autor e as tags da citação. Cada citação extraída é armazenada como um dicionário na lista quotes , organizando os dados para uso ou exportação posterior.

Ótimo! A lógica de raspagem agora está implementada.

Passo #6: Exporte os Dados Raspados

Use essas linhas de código para exportar os dados extraídos para um arquivo 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)

O trecho acima abre um arquivo chamado quotes.csv no modo de gravação. Em seguida, configura os cabeçalhos das colunas (text, author, tags), grava os cabeçalhos e, em seguida, grava cada dicionário da lista quotes no arquivo CSV.

csv.DictWriter simplifica a formatação de dados, facilitando o armazenamento de dados estruturados. Para fazer isso funcionar, lembre-se de importar csv da Biblioteca Padrão do Python:

import csv

Etapa 7: Junte Tudo

É assim que seu script final de web scraping AIOHTTP deve ficar:

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

Você pode executá-lo com:

python scraper.py

Ou, no Linux/macOS:

python3 scraper.py

O arquivo quotes.csv aparecerá na pasta raiz do seu projeto. Abra e você verá:

E pronto! Você acabou de aprender a fazer web scraping com AIOHTTP e BeautifulSoup.

AIOHTTP para raspagem de dados na Web: recursos e técnicas avançadas

Agora que você entende como usar o AIOHTTP para extração básica de dados da web, é hora de ver cenários mais avançados.

Nos exemplos a seguir, o site de destino será o HTTPBin.io /anything endpoint. Essa é uma API útil que retorna o endereço IP, cabeçalhos e outros dados enviados pelo solicitante.

Prepare-se para dominar o AIOHTTP para web scraping!

Definir Cabeçalhos Personalizados

Você pode especificar cabeçalhos personalizados em uma solicitação AIOHTTP usando o 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())

Dessa forma, o AIOHTTP fará uma solicitação GET HTTP com os cabeçalhos Accept e Accept-Language definidos.

Definir um agente de usuário personalizado

User-Agent é um dos cabeçalhos HTTP mais críticos para web scraping. Por padrão, o AIOHTTP usa este User-Agent:

Python/<PYTHON_VERSION> aiohttp/<AIOHTTP_VERSION>

O valor padrão acima pode facilmente expor suas solicitações como provenientes de um script automatizado. Isso aumentará o risco de ser bloqueado pelo site de destino.

Para reduzir as chances de ser detectado, você pode definir um User-Agent personalizado do 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())

Descubra os melhores agentes de usuário para web scraping!

Definir cookies

Assim como os cabeçalhos HTTP, você pode definir cookies personalizados usando os cookies em 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())

Os cookies ajudam você a incluir os dados da sessão necessários em suas solicitações de web scraping.

Observe que os cookies definidos em ClientSession são compartilhados em todas as solicitações feitas com essa sessão. Para acessar os cookies de sessão, consulte ClientSession.cookie_jar.

Integração de proxy

No AIOHTTP, você pode rotear suas solicitações por meio de um servidor proxy para reduzir o risco de proibições de IP. Faça isso usando o proxy argumento na função do método HTTP na sessão :

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

Descubra como realizar a autenticação e rotação de proxy em nosso guia sobre como usar um proxy no AIOHTTP.

Tratamento de erros

Por padrão, o AIOHTTP gera erros somente para problemas de conexão ou rede. Para gerar exceções para respostas HTTP ao receber os códigos de status 4xx e 5xx , você pode usar qualquer uma das seguintes abordagens:

1. Defina raise_for_status=True ao criar a ClientSession: Levante automaticamente exceções para todas as solicitações feitas por meio da sessão se o status da resposta for 4xx ou 5xx.
2. Pass raise_for_status=True diretamente para os métodos de solicitação: Ative a geração de erros para métodos de solicitação individuais (como session.get() ou session.post()) sem afetar os outros.
3. Chame response.raise_for_status() manualmente: Dê controle total sobre quando lançar exceções, permitindo que você decida para cada solicitação.

Exemplo da opção #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())

Quando raise_for_status=True é definido no nível da sessão, todas as solicitações feitas por meio dessa sessão gerarão um aiohttp.ClientResponseError para 4xx ou 5xx respostas.

Exemplo da opção #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())

Nesse caso, o argumento raise_for_status=True é passado diretamente para a chamada session.get() . Isso garante que uma exceção seja gerada automaticamente para qualquer código de status 4xx ou 5xx .

Exemplo da opção #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())

Se você preferir mais controle sobre solicitações individuais, você pode chamar response.raise_for_status() manualmente após fazer uma solicitação. Essa abordagem permite que você decida exatamente quando lidar com os erros.

Tentar novamente solicitações que falharam

O AIOHTTP não oferece suporte nativo para re-tentar solicitações automaticamente. Para implementar isso, você deve usar uma lógica personalizada ou uma biblioteca de terceiros como aiohttp-retry. Isso permite que você configure a lógica de repetição para solicitações com falha, ajudando a lidar com problemas transitórios de rede, tempos limite ou limites de taxa.

Instale aiohttp-retry com:

pip install aiohttp-retry

Em seguida, você pode usá-lo da seguinte forma:

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

Isso configura o comportamento de re-tentativas, com uma estratégia de recuo exponencial. Saiba mais nos documentos oficiais.

AIOHTTP vs Requests para Web Scraping

Abaixo está uma tabela de resumo para comparar AIOHTTP e Requests para web scraping:

Para uma comparação completa, confira nossa postagem no blog sobre Requests versus HTTPX versus AIOHTTP.

Saiba como raspar sites com HTTPX.

Conclusão

Neste artigo, você aprendeu como usar a aiohttp biblioteca para raspagem de dados na web. Você explorou o que ele é, os recursos que ele oferece e os benefícios que ele proporciona. O AIOHTTP se destaca como uma opção rápida e confiável para fazer solicitações HTTP ao coletar dados on-line.

No entanto, solicitações HTTP automatizadas expõem seu endereço IP público. Isso pode revelar sua identidade e localização, colocando sua privacidade em risco. Para proteger sua segurança e privacidade, uma das estratégias mais eficazes é usar um servidor proxy para ocultar seu endereço IP.

A Bright Data controla os melhores servidores de proxy do mundo, atendendo a empresas listadas na Fortune 500 e a mais de 20.000 clientes. Sua oferta inclui uma grande variedade de tipos de proxy:

• Proxies de data centers — mais de 770.000 IPs de data centers.
• Proxies residenciais — mais de 72 milhões de IPs residenciais em mais de 195 países.
• Proxies de ISP — mais de 700.000 IPs de ISP.
• Proxies móveis – mais de 7 milhões de IPs móveis.

Crie uma conta gratuita da Bright Data hoje para testar nossos proxies e soluções de raspagem!