このガイドでは、以下の内容について説明します。
- AIOHTTPとは何か、そしてAIOHTTPが提供する主な機能
- AIOHTTPを使用したウェブスクレイピングに関するステップバイステップのセクション
- AIOHTTPによるウェブスクレイピングの高度なテクニック
- 自動リクエストの処理に関するAIOHTTPとRequestsの比較
さっそく始めましょう！
AIOHTTPとは何ですか？
AIOHTTPは、Pythonのasyncio上に構築された非同期クライアント/サーバーHTTPフレームワークです。従来のHTTPクライアントとは異なり、AIOHTTPはクライアントセッションを使用して複数のリクエストにわたって接続を維持します。そのため、高並行処理性が求められるセッションベースのタスクには効率的な選択肢となります。
⚙️機能
- HTTPプロトコルのクライアントおよびサーバーの両方をサポート。
- WebSocket (クライアントおよびサーバーの両方)のネイティブサポートを提供。
- Webサーバーにミドルウェアとプラガブルなルーティングを提供。
- 大容量データのストリーミングを効率的に処理。
- クライアントセッション永続性により、接続の再利用が可能になり、複数のリクエストのオーバーヘッドを軽減。
AIOHTTPを使用したスクレイピング: ステップバイステップのチュートリアル
ウェブスクレイピングのコンテキストでは、AIOHTTPはページの生のHTMLコンテンツを取得するためのHTTPクライアントに過ぎません。そのHTMLからデータを解析して抽出するには、BeautifulSoupのようなHTMLパーサーが必要です。
このセクションでは、BeautifulSoupでのウェブスクレイピングにAIOHTTPを使用する方法を学習します。
警告: AIOHTTPは主にプロセスの初期段階で使用されますが、スクレイピングのワークフロー全体を紹介します。より高度なAIOHTTPウェブスクレイピング技術に興味がある方は、ステップ3の後の章に進んでください。
ステップ#1:スクレイピングプロジェクトの設定
コンピュータにPython 3+がインストールされていることを確認してください。インストールされていない場合は、公式サイトからダウンロードし、インストール手順に従ってください。
次に、以下のコマンドを使用してAIOHTTPスクレイピングプロジェクト用のディレクトリを作成します。
mkdir aiohttp-scraper
そのディレクトリに移動して仮想環境をセットアップします。
cd aiohttp-scraper
python -m venv env
任意のPython IDEでプロジェクトフォルダを開きます。 Python ExtensionがインストールされたVisual Studio CodeおよびPyCharm Community Editionのいずれも使用できます。
次に、プロジェクトフォルダ内にscraper.pyファイルを作成します。このファイルは最初は空ですが、すぐにスクレイピングロジックを追加します。
IDEのターミナルで、仮想環境を有効にします。LinuxまたはmacOSの場合は、以下を使用してください。
./env/bin/activate
同様に、Windowsの場合は以下を実行します。
env/Scripts/activate
準備ができました！以上で設定は完了で、利用の準備が整いました。
ステップ#2:スクレイピングライブラリの設定
仮想環境が有効な状態で、以下のコマンドを使用してAIOHTTPとBeautifulSoupをインストールします。
pip install aiohttp beautifulsoup4
これでaiohttpとbeautifulsoup4の両方がプロジェクトの依存関係に追加されます。
それらをscraper.pyスクリプトにインポートします。
import asyncio
import aiohttp
from bs4 import BeautifulSoup
aiohttpが機能するためにはasyncioが必要であることに注意してください。
次に、以下のasync関数ワークフローをscrper.pyファイルに追加します。
async def scrape_quotes():
# Scraping logic...
# Run the asynchronous function
asyncio.run(scrape_quotes())
scrape_quotes ()は、スクレイピングロジックがブロックされずに並行して実行される非同期関数を定義します。最後に、asyncio.run (scrape_quotes ())が非同期関数を起動して実行します。
完了です！これで、スクレイピングワークフローの次のステップに進むことができます。
ステップ#3:ターゲットページのHTMLの取得
この例では、「Quotes to Scrape」サイトからデータをスクレイピングする方法を説明します。
RequestsやAIOHTTPのようなライブラリでは、GETリクエストを行うだけでページのHTMLコンテンツを直接入手することができます。ただし、AIOHTTPは異なるリクエストライフサイクルに従います。
AIOHTTPのプライマリコンポーネントはClientSessionで、これは接続のプールを管理し、デフォルトでKeep-Aliveをサポートします。リクエストのたびに新しい接続を開く代わりに、接続を再利用してパフォーマンスを向上させます。
リクエストを行う場合、通常、プロセスには次の3つのステップが含まれます。
ClientSession ()を使ってセッションを開く。session.get ()を使用してGETリクエストを非同期で送信する。await response.text ()などのメソッドを使用して応答データにアクセスする。
この設計により、操作間で異なるwithコンテキストを使用しながらもブロックせずにイベントループを実行できるため、高並行性のタスクに最適です。
したがって、このロジックを使用してAIOHTTPでホームページのHTMLを取得できます。
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()
バックグラウンドで、AIOHTTPはリクエストをサーバーに送信し、ページのHTMLを含む応答を待機します。応答が受信されると、await response.text ()がHTMLコンテンツを文字列として抽出します。
html変数を出力すると、以下のようになります。
<!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>
その調子です！このように、ターゲットページのHTMLコンテンツが正常に取得されました。次に、このコンテンツを解析して、必要なデータを抽出します。
ステップ#4: HTMLドキュメントの解析
以下のように、HTMLコンテンツをBeautifulSoupコンストラクタに渡して解析します。
# Parse the HTML content using BeautifulSoup
soup = BeautifulSoup(html, "html.parser")
html.parserは、コンテンツの処理に使用されるデフォルトのPython HTMLパーサーです。
soupオブジェクトには解析済みのHTMLが含まれ、必要なデータを抽出するメソッドを提供します。
AIOHTTPがHTMLの取得を処理したので、今度はBeautifulSoupを使用した一般的なデータ解析フェーズに移行します。詳細については、Beautiful Soupウェブスクレイピングのチュートリアルをご覧ください。
ステップ#5:データ抽出ロジックを書く
以下のコードを使用してページから引用データを抽出できます。
# 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
})
このスニペットは、スクレイピングされたデータを格納するquotesという名前のリストを初期化します。次に、すべての引用HTML要素を識別し、それらをループ処理して引用テキスト、作成者、およびタグを抽出します。抽出された各引用はquotesリストに辞書として保存され、後で使用またはエクスポートできるようにデータを整理します。
素晴らしい！これで、スクレイピングロジックが実装されました。
ステップ#6:スクレイピングしたデータをJSONにエクスポートする
以下のコードを使って、スクレイピングしたデータを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)
上記のスニペットはquotes.csvという名前のファイルを書き込みモードで開きます。次に、このスニペットは列ヘッダー(テキスト、作成者、タグ)を設定し、ヘッダーを書き込み、quotesリストの各辞書をCSVファイルに書き込みます。
csv.dictWriterは、データのフォーマットを簡素化し、構造化データを簡単に保存できるようにします。これを機能させるには、Python標準ライブラリからcsvをインポートすることを忘れないでください。
import csv
ステップ#7:仕上げ
最終的な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())
以下で実行できます。
python scraper.py
または、Linux/macOSの場合:
python3 scraper.py
quotes.csvファイルがプロジェクトのルートフォルダに表示されます。これを開くと、次の内容が表示されます。
ほら、このとおり！ここまで、AIOHTTPとBeautifulSoupを使ってウェブスクレイピングを実行する方法を学びました。
ウェブスクレイピング用AIOHTTP:高度な機能とテクニック
基本的なウェブスクレイピングにAIOHTTPを使用する方法を理解できたので、ここからはさらに高度なシナリオを見てみましょう。
以下の例では、ターゲットサイトはHttpBin.io /anythingエンドポイントです。これは、リクエスタが送信したIPアドレス、ヘッダー、その他のデータを返す便利なAPIです。
ウェブスクレイピング用のAIOHTTPをマスターしましょう！
カスタムヘッダーの設定
AIOHTTPリクエストwithでヘッダー引数にカスタムヘッダーを指定できます。
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())
こうすることで、AIOHTTPはAcceptとAccept-Languageヘッダーが設定されたGET HTTPリクエストを作成します。
カスタムユーザーエージェントの設定
ユーザーエージェントはウェブスクレイピングにおいて最も重要なHTTPヘッダーのひとつです。デフォルトでは、AIOHTTPはこのユーザーエージェントを使用します。
Python/<PYTHON_VERSION> aiohttp/<AIOHTTP_VERSION>
上記のデフォルト値では、リクエストが自動スクリプトからのものであることが簡単にわかってしまいます。よって、ターゲットサイトによってブロックされるリスクが高まります。
検出される可能性を減らすために、以前と同様にカスタムの現実世界のユーザーエージェントを設定できます。
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())
ウェブスクレイピングに最適なユーザーエージェントをご覧ください！
クッキーの設定
HTTPヘッダーと同様に、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())
クッキーを使用すると、ウェブスクレイピングリクエストに必要なセッションデータを含めることができます。
ClientSessionに設定されたクッキーは、そのセッションで行われたすべてのリクエストで共有されることに注意してください。セッションクッキーへのアクセスについては、ClientSession.cookie_jarを参照してください。
プロキシの統合
AIOHTTPでは、IP制限のリスクを軽減するために、リクエストをプロキシサーバー経由でルーティングできます。そのためには、セッションのHTTPメソッド関数でプロキシ引数を使用します:
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())
プロキシ認証とローテーションの実行方法については、AIOHTTPでプロキシを使用する方法に関するガイドをご覧ください。
エラー処理
デフォルトでは、AIOHTTPは接続またはネットワークの問題のみでエラーを発生します。4xxおよび5xxステータスコードを受信したときにHTTP応答の例外を発生させるには、以下のいずれかの方法を使用できます。
ClientSessionの作成時にraise_for_status=Trueを設定する。これにより、応答ステータスが4xxまたは5xxの場合、セッションを通じて行われたすべてのリクエストの例外を自動的に発生させます。raise_for_status=Trueをリクエストメソッドに直接渡す。これにより、個々のリクエストメソッド (session.get ()やsession.post ()など)で他のリクエストメソッドに影響を及ぼすことなくエラーを発生させることができます。response.raise_for_status ()を手動で呼び出す。これにより、例外を発生させるタイミングを完全に制御できるため、リクエストごとに決定できます。
オプション#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())
raise_for_status=Trueがセッションレベルで設定されている場合、そのセッションを通じて行われたすべてのリクエストは、4xxまたは5xxの応答に対してaioHttp.ClientResponseErrorを発生させます。
オプション#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())
この場合、raise_for_status=True引数はsession.get ()呼び出しに直接渡されます。これにより、あらゆる4xxまたは5xxステータスコードで例外が自動的に発生することが保証されます。
オプション#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())
個々のリクエストをより細かく制御したい場合は、リクエストを行った後にresponse.raise_for_status ()を手動で呼び出すことができます。この方法では、エラーを処理するタイミングを正確に決定できます。
失敗したリクエストの再試行
AIOHTTPには、リクエストの自動再試行機能が組み込まれていません。これを実装するには、カスタムロジックまたはaiohttp-retryのようなサードパーティ製ライブラリを使用する必要があります。これにより、失敗したリクエストの再試行ロジックを設定できるため、一時的なネットワークの問題、タイムアウト、またはレート制限の処理に役立ちます。
以下を使ってaiohttp-retryをインストールします。
pip install aiohttp-retry
これで、以下のように使用できます。
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()
これにより、指数バックオフ戦略による再試行動作が設定されます。詳細は、公式ドキュメントをご覧ください。
AIOHTTPおよびRequestsによるウェブスクレイピングの比較
以下は、AIOHTTP と ウェブスクレイピングのリクエストを比較するための要約表です。
| 機能 | AIOHTTP | Requests |
|---|---|---|
| GitHub Star | 15.3k | 52.4k |
| クライアントサポート | ✔️ | ✔️ |
| 同期サポート | ❌ | ✔️ |
| 非同期サポート | ✔️ | ❌ |
| サーバーサポート | ✔️ | ❌ |
| コネクションプーリング | ✔️ | ✔️ |
| HTTP/2サポート | ❌ | ❌ |
| ユーザーエージェントのカスタマイズ | ✔️ | ✔️ |
| プロキシサポート | ✔️ | ✔️ |
| クッキー処理 | ✔️ | ✔️ |
| 再試行処理 | サードパーティ製ライブラリでのみ利用可能 | HTTPAdaptersで使用可能 |
| パフォーマンス | 高 | 中 |
| コミュニティサポートと人気 | 中 | 大 |
総合的な比較については、Requests、HTTPXおよびAIOHTTPに関するブログ記事をご覧ください。
HTTPXを使ってウェブサイトをスクレイピングする方法をご覧ください。
まとめ
この記事では、aiohttpライブラリをウェブスクレイピングに使用する方法をご紹介しました。それがどういうものか、それが提供する機能、そしてそれがもたらす利点を学習しました。AIOHTTPは、オンラインデータを収集するためにHTTPリクエストを行う上で、高速かつ信頼性の高い選択肢として際立っています。
ただし、自動HTTPリクエストではお使いのパブリックIPアドレスが公開されてしまいます。これにより、あなたの身元と場所が明らかになり、プライバシーが危険にさらされる可能性があります。セキュリティとプライバシーを保護するための最も効果的な戦略のひとつは、プロキシサーバーを使用してIPアドレスを隠すことです。
Bright Dataは世界最高レベルのプロキシサーバーを保有しており、フォーチュン500企業を含む20,000社以上の顧客にサービスを提供しています。Bright Dataは次のような様々な種類のプロキシを提供しています。
- データセンタープロキシ — 77万件を超えるデータセンターIP
- 住宅用プロキシ — 195ヵ国以上、7200万件を超える住宅用IP
- ISPプロキシ — 70万件を超えるISP用IP
- モバイルプロキシ — 700万件を超えるモバイル用IP
今すぐ無料のBright Dataアカウントを作成して、プロキシおよびスクレイピングソリューションをテストしてください！
クレジットカードは必要ありません
