Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from functools import wraps
import requests
from requests import Session, ConnectionError
from requests.exceptions import MissingSchema
from requests.packages.urllib3.exceptions import ProtocolError, LocationParseError
from socket import error as SocketError
import errno
import time
def demands(*dargs, **dkwargs):
# set defaults
retry = dkwargs.get('retry', 5)
wait = dkwargs.get('wait', 1)
backoff = dkwargs.get('backoff', 1.25)
status_codes = dkwargs.get('status_codes', [200])
raise_uncaught_errors = dkwargs.get('raise_uncaught_errors', False)
# wrapper
def wrapper(f):
@wraps(f)
def wrapped_func(*args, **kw):
# defaults
r = None
tries = 0
err = True
# for ref problems
_backoff = backoff * 1
_wait = wait * 1
while 1:
# increment tries
tries += 1
# if we've exceeded the maximum number of tries,
# return
if tries > retry:
print "Reached max tries."
return r
# calc wait time for this step
_wait *= _backoff
# try the function
try:
r = f(*args, **kw)
err = False
except ConnectionError:
time.sleep(_wait)
except ProtocolError:
time.sleep(_wait)
except SocketError:
time.sleep(_wait)
except LocationParseError:
time.sleep(_wait)
except MissingSchema:
time.sleep(_wait)
except Exception as e:
if raise_uncaught_errors:
raise e
else:
time.sleep(_wait)
# check the status code
if not err and r and r.status_code in status_codes:
break
elif not err:
time.sleep(_wait)
else:
err = True
return r
return wrapped_func
return wrapper
@demands(retry=5, status_codes=[200], wait=0.7, backoff=1.10, raise_uncaught_errors=False)
def get_url(url):
return requests.get(url)
if __name__ == '__main__':
r = get_url(url="http://www.google.com/")
print r
r = get_url(url="http://google.bar/")
print r
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.