이전에 이런 종류의 오류를 다루었을 때, 주어진 예외가 발생하면 특정 횟수만큼 함수 호출을 다시 시도하는 데코레이터를 작성했습니다.
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 초 후, 세 번째 재시도는 4 초 후 등
(210)
그리고 당신의 재시도 장식으로 장식 된 웹 사이트, 긁어 함수를 정의 : 쉽게 재 시도를 트리거하는 예외를 설정할 수 있습니다
@retry
def scrape_a_site(url, params):
r = requests.get(url, params=params)
return r
참고. 여기에 RequestException
및 timeout
을 사용했습니다. 귀하의 상황에 적응하십시오.
#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" })
주 당신의 작은 블록에 @retry
을 적용 :
코드와 관련하여
, 당신은 (위의 코드의 첫 번째 블록으로 장식을 규정 한) 이런 식으로 뭔가에 적응 할 수 코드 수 (그냥 웹 조회 논리).
때때로 서버의 시간 제한이 30 초로 제한되어 있습니다. 이는 시간 제한 내에 스크립트가 완료되지 않으면 스크립트가 종료된다는 것을 의미합니다. 이것이 이유 인 경우 셀러리를 사용하여 백그라운드 작업에서이 근근이 살아가는 것을 사용할 수 있습니다. –
나는 셀러리를 전혀 사용하지 않았고 리눅스 + 파이썬에서 완전히 초보자이기 때문에 try-except 메소드로 해결할 수 없다면 작업 스케줄링 (crom)으로 스크립트를 다시 열어 보자. –