Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if error_codes.get(req.status_code):
# check to see whether its 429
if req.status_code == 429:
# try to get backoff duration
delay = req.headers.get("backoff")
if not delay:
raise ze.TooManyRetries(
"You are being rate-limited and no backoff duration has been received from the server. Try again later"
)
else:
zot._set_backoff(delay)
else:
raise error_codes.get(req.status_code)(err_msg(req))
else:
raise ze.HTTPError(err_msg(req))
if delay > 32:
# we've waited a total of 62 seconds (2 + 4 … + 32), so give up
backoff.reset()
raise ze.TooManyRetries("Continuing to receive HTTP 429 \
responses after 62 seconds. You are being rate-limited, try again later")
time.sleep(delay)
sess = requests.Session()
new_req = sess.send(req.request)
try:
new_req.raise_for_status()
except requests.exceptions.HTTPError:
error_handler(new_req)
else:
raise error_codes.get(req.status_code)(err_msg(req))
else:
raise ze.HTTPError(err_msg(req))
def retry_zotero(wrapped, _instance, args, kwargs):
"""
Retry the wrapped function if the Zotero API call fails.
Caution: This decorator should only be used on simple functions, as the
whole function is called repeatedly as long a Zotero fails.
"""
attempts = 1
while True:
try:
return wrapped(*args, **kwargs)
except (
requests.exceptions.ConnectionError,
zotero_errors.HTTPError,
zotero_errors.UnsupportedParams
) as e:
current_app.logger.exception(e)
if attempts < current_app.config['KERKO_ZOTERO_MAX_ATTEMPTS']:
current_app.logger.warning(
"The Zotero API call has failed in {func}. "
"New attempt in {wait} seconds...".format(
func=wrapped.__name__,
wait=current_app.config['KERKO_ZOTERO_WAIT']
)
)
attempts += 1
sleep(current_app.config['KERKO_ZOTERO_WAIT'])
else:
current_app.logger.error(
"The maximum number of API call attempts to Zotero has "