山傘のプログラミング勉強日記

プログラミングに関する日記とどうでもよい雑記からなるブログです。

Pythonクローリング&スクレイピング[増補改訂版] ―データ収集・解析のための実践開発ガイドー その8

第4章 実用のためのメソッド

ステータスコードに応じた処理

クローリングを行う上で Web サーバーに負荷をかけすぎないようにクローラーを設計する必要があります。HTTP 通信におけるエラーによって対処する方法が変わります。

import time
import requests

# HTTPステータスコード
# 408: Request Timeout, 500: Internal Server Error, 502: Bad Gateway
# 503: Service Unavailable, 504: GateWay Timeout
TEMPORARY_ERROR_CODE = (408, 500, 502, 503, 504)

def main():
  """
  ステータスコードに応じたエラー処理を行う
  """
  # http://httpbin.org は HTTPクライアントテストに使われる
  response = fetch('http://httpbin.org/status/200,404,503')
  if 200 <= response.status_code < 300:
    print('Success')
  else :
    print('Error!')

def fetch(url: str) -> requests.Request:
  """
  3回のリトライでResponseオブジェクトを返す
  """
  max_retries = 3
  temp_retiries = 0
  while True:
    try:
      print(f'Retrieving {url}...')
      response = requests.get(url)
      print(f'Status:{response.status_code}')
      if response.status_code not in TEMPORARY_ERROR_CODE:
        return response

    except requests.exceptions.RequestException as ex:
      # ネットワークレベルのエラー
      print(f'Network-level exception occured: {ex}')
    
    temp_retiries += 1
    if temp_retiries >= max_retries:
      raise Exception('リトライの上限を超えました')
    
    wait = 2**(temp_retiries - 1)
    print(f'Waiting {wait} seconds...')
    time.sleep(wait)


if __name__ == '__main__':
  main()

tenacity によるリトライ処理

Tenacity — Tenacity documentation

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

# HTTPステータスコード
# 408: Request Timeout, 500: Internal Server Error, 502: Bad Gateway
# 503: Service Unavailable, 504: GateWay Timeout
TEMPORARY_ERROR_CODE = (408, 500, 502, 503, 504)

def main():
  """
  ステータスコードに応じたエラー処理を行う11
  """
  # http://httpbin.org は HTTPクライアントテストに使われる
  response = fetch('http://httpbin.org/status/200,404,503')
  if 200 <= response.status_code < 300:
    print('Success')
  else :
    print('Error!')


@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))
def fetch(url: str) -> requests.Response:
  print(f'Retrieving {url}...')
  response = requests.get(url)
  print(f'Status: {response.status_code}')
  
  if response.status_code not in TEMPORARY_ERROR_CODE:
    return response

  raise Exception(f'Temporary Error: {response.status_code}')

if __name__ == '__main__':
  main()