目次

記事の概略

実行ファイル (GitHubリンク)
バックナンバー
前置き

非同期関数を用いて指定したurlからデータを取得する

指定したurlからデータを取得する方法
非同期でデータを取得するメリット
非同期処理でwebへアクセスする方法

asyncio + aiohttpの実装

同期処理実装例（おまけ）

urllib.request を使用した例
requests を使用した例
selenium を使用した例

記事の概略

非同期処理で指定したurlからデータを取得する方法は複数ある。本記事はasyncio + aiohttpによる実装を行っている
非同期処理のメリットの1つは、時間のかかるタスクを分離させ、並列に処理することが可能な点
非同期処理と同期処理との処理速度の違いを明らかにするため、同期処理によって同じ目的を達成する実装も行ってみたところ、予想通り非同期処理の方が高速に動くことがわかった

実行ファイル

記事と同じ内容を含むnotebookをgithubにアップしました。実際に手を動かしながら実行される際に利用してみてください。

github: https://github.com/Tsucreator/learn_asynchronous

バックナンバー

前置き

内容に誤りや読者へ誤解を招く記載がありましたら、後学のためご指摘いただけるととても励みとなります。有識者の方よろしくお願いします。

非同期関数を用いて指定したurlからデータを取得する

非同期処理の学習のため、実際に動くコードを書きながら理解を深めました。"指定したurlからデータを取る"という切り口で整理してみましたので、ご参考まで。

指定したurlからデータを取得する方法

同期的処理でデータを取得する方法は複数あります。有名なものは以下の通りでしょうか（本記事の下方に実際にQiita, Zennへアクセスしてデータを取得する処理を記載しました。ご興味のある方は触ってみてください）。

urllib.request: 指定されたURLを開きHTTPレスポンスオブジェクトを取得するためのモジュール。pythonにおけるリクエストとレスポンスのデータハンドリングでは基本的な機能
request: urllib.requestと同等の機能を有するモジュール。urllibと異なり、GETリスクエストをgetメソッドで実行可能など、直感的な操作が可能である
Selenium: Webブラウザを自動的に操作するフレームワーク。ブラウザ操作が可能なのでJSによって動的にコンテンツが生成されるwebからのデータ取得などで利用する
Scrapy: 大規模なWebコンテンツのクローリングプロジェクトに適したフレームワーク。効率的なデータ抽出、データ処理、データ保存の機能を提供

非同期でデータを取得するメリット

非同期のメリットは、時間がかかる処理で全体の処理が止まらない点です。
先述の方法でも困らないとき、非同期処理を利用する必要はありません。web上にある特定のデータを収集するのみ、などでしたら積極的に非同期処理を使用することは少ないと考えられます。
一方で、データを取得した後に続けて処理が行われるときは、同期処理ではユーザー応答に時間がかかる可能性が生じてきます（例えば、後述のSeleniumによるデータ取得を実行していただくとかなり時間がかかることを体感できます）。
urlへアクセスしデータを取得する一連のプロセスを非同期処理として分離することで応答速度を向上させることができます。

非同期処理でwebへアクセスする方法

非同期的な処理で指定したurlよりデータを取得する方法もいくつかあります。有名なものは以下の通りでしょうか

asyncio + aiohttp: 非同期処理の基本的な構成で高いパフォーマンスを発揮する
asyncio + httpx: 高いパフォーマンスを発揮する。またhttpxはrequestsのAPIと互換性があるため同期処理からの移行時に便利
Grequests: 同様にrequestsのAPIと互換性のある、requestsライブラリをベースとした非同期HTTPリクエストライブラリ。グリーンスレッド（OSと異なる仮想的な空間で動作するスレッド）を利用した軽量な処理が可能。ノンブロッキングI/O（I/O処理が完了するまでプログラムの実行が停止しない仕組み）が可能
Twisted: 非同期ネットワークプログラミングのためのフレームワーク。歴史が長く、安定しており、幅広いプロトコル（TCP, UDP, SSL/TLSなど）をサポート。チャットサーバーやゲームサーバーなど、多様なアプリケーションの開発に用いられている

`asyncio + aiohttp`の実装

今回は一例としてasyncio + aiohttpで非同期処理を実装してみたいと思います。（気が向いたら他の方法での実装もやってみます）
Qiitaサイト、Zennサイトアクセスしホーム画面にpostされている記事を取得する処理を実装してみます。非同期処理を使用せずに処理行った時との比較は後述の"同期処理実装例（おまけ）"を触ってみてください。簡単にコードの説明も記載しておきます。

aiohttp.ClientSession() as session: *HTTPクライアントとしての機能を提供するオブジェクト。Session情報を管理する
asyncio.gather(): 引数に渡されたタスクを並列処理するメソッド

※ HTTPクライアントとは、HTTP (Hypertext Transfer Protocol) を使用して、HTTPサーバーにリクエストを送信し、レスポンスを受信するソフトウェアまたはライブラリのこと

import asyncio
import aiohttp

# 非同期で対象urlのhtmlデータを取得する
async def fetch_data(url):
   # Session情報を管理するオブジェクトを作成
   async with aiohttp.ClientSession() as session:
       # GETリクエストでresponseを取得
       async with session.get(url) as response:
           # textメソッドで取得したresponseのContent-Typeを読み取り自動でデコードを行う
           return await response.text()

async def main():
   # 取得したHTMLからaタグのurlをパースするための正規表現（htmlの中身によって調整は必要）
   qiita_url_match = re.compile(r"(?<=a href\=\")https://qiita.com/.*?(?=\")") # Qiita
   zenn_url_match = re.compile(r"(?<=href\=\")[^>]*articles[^>]*?(?=\")") # Zenn
   # 各記事のtitleタグをパースするための正規表現
   title_match = re.compile(r"(?<=\<title\>).*?(?=\<\/title\>)")
   # 取得先のurlをリストで指定
   urls = [
       r"https://qiita.com/",
       r"https://zenn.dev/"
   ]

   # 非同期処理用のタスクを作成し、イベントループに登録
   tasks = [fetch_data(url) for url in urls]
   # イベントループ内のタスクを並列処理で実行
   results = await asyncio.gather(*tasks)
   # 取得したurlを格納する箱
   qiita_match_urls = []
   zenn_match_urls = []
   # 各タスクの結果からresultを取得して処理
   for result in results:
       # print(result) # 結果確認
       url_match = qiita_url_match if "qiita" in result else zenn_url_match
       # 文字列を一行ごとに分割
       html_lines = result.splitlines()
       for line in html_lines:
         if "qiita" in result:
           qiita_match_urls += url_match.findall(line)
         else:
           zenn_match_urls += url_match.findall(line)
   # 集合を用いて重複削除（画面に描画されている記事を上から5つ分）
   qiita_match_urls = list(sorted(set(qiita_match_urls), key=qiita_match_urls.index))[:5]
   zenn_match_urls = [str(r"https://zenn.dev" + url) for url in list(sorted(set(zenn_match_urls), key=zenn_match_urls.index))][:5]

   # Postされている記事のurlを再度格納
   urls = qiita_match_urls + zenn_match_urls
   # 非同期処理用のタスクを作成し、イベントループに登録
   tasks = [fetch_data(url) for url in urls]
   article_titles = await asyncio.gather(*tasks)
   # print(article_titles) # 結果確認
   # 各記事のタイトルを取得するため各urlのtitleを抽出（上位5つ)
   for i, url in enumerate(urls):
     print(title_match.findall(article_titles[i])[0], ": ", url)

# 実行
if __name__ == "__main__":
 # jupyter notebook上では、asyncio.runでは動かないためawaitで実行
 await main()

同期処理実装例（比較用）

以降は、非同期処理に対する同期処理の処理時間を比較するために実装してみました。どれくらい時間が変化するのを体感するためにもぜひ触ってみてください。

import urllib.request
import re

'''qiita, zennにpostされている記事のリンクを取得する'''
def fetch_data(url: str, pattern: re):
  try:
      # requestによって取得したHTTPレスポンスを"response"として保持
      with urllib.request.urlopen(url) as response:
          # read()でhtmlを読み込む
          html = response.read()
          # 取得したデータ(bytes型)をデコードして文字列にする
          html_str = html.decode("utf-8")  # または "shift-jis" など、適切なエンコーディングを指定
          # print(html_str) # 結果確認
          # 文字列を一行ごとに分割
          html_lines = html_str.splitlines()
          # HTML一行ごとに情報を抜いてくる
          match_items = []
          for line in html_lines:
            match_items += pattern.findall(line)
            if not(match_items):
              continue
          # 集合を用いて重複削除（画面に描画されている記事を上から5つ分）
          # zennの場合はhrefで抽出したurlに加筆
          if url == r"https://zenn.dev/":
            match_items = [str(r"https://zenn.dev" + item) for item in match_items]
          return list(sorted(set(match_items), key=match_items.index))[:5]

  except urllib.error.URLError as e:
      print(f"URLエラー: {e.reason}")
  except urllib.error.HTTPError as e:
      print(f"HTTPエラー: {e.code}")
  except Exception as e:
      print(f"エラーが発生しました: {e}")

# 取得先のurlをリストで指定
urls = [
    r"https://qiita.com/",
    r"https://zenn.dev/"
]
# 取得したHTMLからaタグのurlをパースするための正規表現（htmlの中身によって調整は必要）
qiita_url_match = re.compile(r"(?<=a href\=\")https://qiita.com/.*?(?=\")") # Qiita
zenn_url_match = re.compile(r"(?<=href\=\")[^>]*articles[^>]*?(?=\")") # Zenn
# 各記事のtitleタグをパースするための正規表現
title_match = re.compile(r"(?<=\<title\>).*?(?=\<\/title\>)")


# 実行
# 取得した記事urlのリスト
article_urls = []
for url, match_pattern in zip(urls, [qiita_url_match, zenn_url_match]):
  article_urls += fetch_data(url, match_pattern)

## 各記事のタイトルを取得するため各urlのtitleを抽出（上位5つずつ）
for url in article_urls:
  article_title = fetch_data(url, title_match)
  print(article_title[0], ": ", url)

※requestによるデータ取得例

import requests
import re

'''qiita, zennにpostされている記事のリンクを取得する'''
def fetch_data(url: str, pattern: re):
  try:
    # GETリクエストでレスポンスを取得。
    response = requests.get(url, timeout=10) # 10秒でタイムアウト
    # HTTPエラーが発生した場合に例外を発生させる処理
    response.raise_for_status()
    # responseのbodyを文字列として取得 (response.textでは自動でデコードする)
    # requestsではレスポンスヘッダーのContent-Typeに含まれるcharset情報に基づいて自動的にエンコーディングを判別しデコードする
    # 文字コードはencodeingプロパティが保持しており、明示的に指示する時にはresponse.encoding = "XXXXXX" と指定する
    html = response.text
    # print(html) # 結果確認
    # 文字列を一行ごとに分割
    html_lines = html.splitlines()
    # HTML一行ごとに情報を抜いてくる
    match_items = []
    for line in html_lines:
      match_items += pattern.findall(line)
      if not(match_items):
        continue
    # 集合を用いて重複削除（画面に描画されている記事を上から5つ分）
    # zennの場合はhrefで抽出したurlに加筆
    if url == r"https://zenn.dev/":
      match_items = [str(r"https://zenn.dev" + item) for item in match_items]
    return list(sorted(set(match_items), key=match_items.index))[:5]

  except requests.exceptions.Timeout:
    print("タイムアウトしました")
  except requests.exceptions.RequestException as e:
    print(f"エラーが発生しました: {e}")

# 取得先のurlをリストで指定
urls = [
    r"https://qiita.com/",
    r"https://zenn.dev/"
]
# 取得したHTMLからaタグのurlをパースするための正規表現（htmlの中身によって調整は必要）
qiita_url_match = re.compile(r"(?<=a href\=\")https://qiita.com/.*?(?=\")") # Qiita
zenn_url_match = re.compile(r"(?<=href\=\")[^>]*articles[^>]*?(?=\")") # Zenn
# 各記事のtitleタグをパースするための正規表現
title_match = re.compile(r"(?<=\<title\>).*?(?=\<\/title\>)")

# 実行
# 取得した記事urlのリスト
article_urls = []
for url, match_pattern in zip(urls, [qiita_url_match, zenn_url_match]):
  article_urls += fetch_data(url, match_pattern)

## 各記事のタイトルを取得するため各urlのtitleを抽出（上位5つずつ）
for url in article_urls:
  article_title = fetch_data(url, title_match)
  print(article_title[0], ": ", url)

※seleniumによるデータ取得例（seleniumのinstallを忘れずに実行してください）

# !pip install selenium # 初回ではseleniumをinstallする
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import re

'''qiita, zennにpostされている記事のリンクを取得する'''
def fetch_data(url: str, pattern: re):
  # Headless Chromeの設定
  # 起動時の設定を管理するoptionsオブジェクトを作成
  options = webdriver.ChromeOptions()
  # ヘッドレスモードで動かすように設定。ヘッドレスモードはGUIなしで動くためバックグラウンドで動かせる
  options.add_argument("--headless")
  # GUIを使用しないので念の為GPUのレンダリングを無効に設定。
  options.add_argument("--disable-gpu")
  # sandbox環境での実行を無効に設定。sandbox環境はブラウザを仮想的な環境で実行する機能。有効にすることでセキュリティが向上するがColab上ではエラーとなるため無効
  options.add_argument("--no-sandbox")

  try:
    # Chromeを起動(optionsは上記の設定を反映するため)
    driver = webdriver.Chrome(options=options)
    # driverオブジェクトのget()メソッドを用いてChrome上でGETリクエストを出す。
    driver.get(url)
    # アクセス先のHTMLを取得するためpage_sourceを取得
    html = driver.page_source
    # print(html) 確認用
    html_lines = html.splitlines()
    # HTML一行ごとに情報を抜いてくる
    match_items = []
    for line in html_lines:
      match_items += pattern.findall(line)
      if not(match_items):
        continue
    # 集合を用いて重複削除（画面に描画されている記事を上から5つ分）
    # zennの場合はhrefで抽出したurlに加筆
    if url == r"https://zenn.dev/":
      match_items = [str(r"https://zenn.dev" + item) for item in match_items]
    return list(sorted(set(match_items), key=match_items.index))[:5]

  except Exception as e:
    print(f"エラーが発生しました: {e}")


# 取得先のurlをリストで指定
urls = [
    r"https://qiita.com/",
    r"https://zenn.dev/"
]
# 取得したHTMLからaタグのurlをパースするための正規表現（htmlの中身によって調整は必要）
qiita_url_match = re.compile(r"(?<=a href\=\")https://qiita.com/.*?(?=\")") # Qiita
zenn_url_match = re.compile(r"(?<=href\=\")[^>]*articles[^>\.js]*?(?=\")") # Zenn
# 各記事のtitleタグをパースするための正規表現
title_match = re.compile(r"(?<=\<title\>).*?(?=\<\/title\>)")

# 実行
# 取得した記事urlのリスト
article_urls = []
for url, match_pattern in zip(urls, [qiita_url_match, zenn_url_match]):
  article_urls += fetch_data(url, match_pattern)
  
## 各記事のタイトルを取得するため各urlのtitleを抽出（上位5つずつ）
for url in article_urls:
  article_title = fetch_data(url, title_match)
  print(article_title[0], ": ", url)