以前私がこれらの種類のエラーに対処してきたときに、特定の例外が発生した場合に関数呼び出しを一定回数再試行するデコレータを作成しました。
from functools import wraps
import time
from requests.exceptions import RequestException
from socket import timeout
class Retry(object):
"""Decorator that retries a function call a number of times, optionally
with particular exceptions triggering a retry, whereas unlisted exceptions
are raised.
:param pause: Number of seconds to pause before retrying
:param retreat: Factor by which to extend pause time each retry
:param max_pause: Maximum time to pause before retry. Overrides pause times
calculated by retreat.
:param cleanup: Function to run if all retries fail. Takes the same
arguments as the decorated function.
"""
def __init__(self, times, exceptions=(IndexError), pause=1, retreat=1,
max_pause=None, cleanup=None):
"""Initiliase all input params"""
self.times = times
self.exceptions = exceptions
self.pause = pause
self.retreat = retreat
self.max_pause = max_pause or (pause * retreat ** times)
self.cleanup = cleanup
def __call__(self, f):
"""
A decorator function to retry a function (ie API call, web query) a
number of times, with optional exceptions under which to retry.
Returns results of a cleanup function if all retries fail.
:return: decorator function.
"""
@wraps(f)
def wrapped_f(*args, **kwargs):
for i in range(self.times):
# Exponential backoff if required and limit to a max pause time
pause = min(self.pause * self.retreat ** i, self.max_pause)
try:
return f(*args, **kwargs)
except self.exceptions:
if self.pause is not None:
time.sleep(pause)
else:
pass
if self.cleanup is not None:
return self.cleanup(*args, **kwargs)
return wrapped_f
は、あなたが機能(最大の再試行の後に)失敗したコールに対処することができます
:
def failed_call(*args, **kwargs):
"""Deal with a failed call within various web service calls.
Will print to a log file with details of failed call.
"""
print("Failed call: " + str(args) + str(kwargs))
# Don't have to raise this here if you don't want to.
# Would be used if you want to do some other try/except error catching.
raise RequestException
は、あなたの関数を飾るために、クラスのインスタンスを作成するには、呼び出します。retreat=2
で
#Class instance to use as a retry decorator
retry = Retry(times=5, pause=1, retreat=2, cleanup=failed_call,
exceptions=(RequestException, timeout))
最初の再試行は1秒後に、2回目の再試行は2秒後、3回目の再試行は4秒後に行われます。
そして、あなたの再試行デコレータで飾られたウェブサイトを、こすりするためにあなたの関数を定義します。あなたは簡単に再試行をトリガする例外を設定することができます
@retry
def scrape_a_site(url, params):
r = requests.get(url, params=params)
return r
注意を。私はここでRequestException
とtimeout
を使用しました。あなたの状況に適応してください。あなたのコードに関して
、あなたは、この(上記のコードの最初のブロックを使用してデコレータを定義した)のようなものに適合させることができます:あなたが最小のブロックに@retry
を適用している
#Class instance to use as a retry decorator
retry = Retry(times=5, pause=1, retreat=2, cleanup=None,
exceptions=(RequestException, timeout))
@retry
def get_html(browser, url):
'''Get HTML from url'''
browser.get(url)
return browser.page_source
def scrape(urls):
browser = webdriver.Firefox()
datatable=[]
for url in urls:
html = get_html(browser, url)
soup=BeautifulSoup(html,"html.parser")
table = soup.find('table', { "class" : "table table-condensed table-hover data-table m-n-t-15" })
注意あなたができるコード(ちょうどウェブルックアップロジック)。
時々、サーバーのタイムアウト時間が30秒に制限されています。タイムアウト時間内にスクリプトが完了しないと、スクリプトが強制終了されることを意味します。それが理由であれば、セーラーを使用してバックグラウンド・タスクでこのスクレイピングを使用することができます。 –
私は決してセロリを使用していませんでした。プラスLinux + Pythonで完全に初心者でしたので、試してみる方法で解決できない場合、私はちょうどジョブスケジューリング(crom)でスクリプトを再開します –