Skip to content

Instantly share code, notes, and snippets.

@abelsonlive
Created September 16, 2014 21:38
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save abelsonlive/13c213ddf9fa744e7dc1 to your computer and use it in GitHub Desktop.
Save abelsonlive/13c213ddf9fa744e7dc1 to your computer and use it in GitHub Desktop.
#!/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