def context(conf, retry = None, fake_error = None): """ Examples: - Do not fail unless the third try goes also wrong: retries = 3 - wait delay_ms between eacht try - Succeed on the 3rd try: fails=2 retry = dict(retries=2, delay_ms=1) fake_error = dict(fails=3) ctx = swapi.context(conf) """ res = dict( conf=conf, retry = dict( retries=3, delay_ms=2000, ), fake_error = dict(), # fake_error = dict(fails=2) ) if retry is not None: res["retry"].update(retry) if fake_error is not None: res["fake_error"].update(fake_error) return res class SwapiFakeIOError(OSError): pass """ This is our Simulated IO Error Exception for Unit Testing. From Python 3.3 on this is the relevant IO Exception hierarchy: +-- OSError | +-- BlockingIOError | +-- ChildProcessError | +-- ConnectionError | | +-- BrokenPipeError | | +-- ConnectionAbortedError | | +-- ConnectionRefusedError | | +-- ConnectionResetError | +-- FileExistsError | +-- FileNotFoundError | +-- InterruptedError | +-- IsADirectoryError | +-- NotADirectoryError | +-- PermissionError | +-- ProcessLookupError | +-- TimeoutError """ def rest_call(method, url, auth, data = None, fake_error = {}): fails_left = fake_error.get("fail_times", 0) if fails_left > 0: msg = "%s %s fails, %s fake errors left." % ( method, url, fails_left - 1) LOG.debug(msg) raise SwapiFakeIOError(msg) import requests if method == "GET": r = requests.get( url = url, auth = auth, ) if method == "POST": r = requests.delete( url = url, auth = auth, ) if method == "PUT": r = requests.delete( url = url, auth = auth, ) if method == "DELETE": r = requests.delete( url = url, auth = auth, ) return r def construct_auth(conf): proto, server, basepath, user, key = conf import requests.auth auth=requests.auth.HTTPDigestAuth(user, key) return auth def construct_url(conf, coll): """ >>> conf = ("http","example.com", "/subshop/", "api", "key123key") >>> construct_url(conf,"articles") 'http://example.com/subshop/api/articles' >>> conf = ("https","example.com", "/", "api", "key123key") >>> construct_url(conf,"orders") 'https://example.com/api/orders' """ proto, server, basepath, user, key = conf url = '%s://%s%sapi/%s' % (proto,server,basepath, coll) import swapi swapi.LOG.debug("Url constructed: %s", url) return url def get(ctx, coll, suffix = ""): #next_action, exception, message, result = handle_context(ctx) #if next_action == "return" # return result #if next_action == "exception" # raise conf = ctx["conf"] url = "%s%s" % (construct_url(conf,coll), suffix) global LOG # use LOG from the module namespace import swapi swapi.LOG.debug("Auth user: %s", conf[3]) auth = construct_auth(conf) import requests last_i = ctx["retry"]["retries"]+1 assert last_i > 0 for i in range(0,last_i): try: r = rest_call( method = "GET", url = url, auth = auth, fake_error = ctx["fake_error"], ) except OSError as e: # all IO Errors are catched here import traceback tb = traceback.format_exc() LOG.error("OSError exception: %s" % tb) # are we out of retries? if i >= ctx["retry"]["retries"]: # then re-raise it! raise except: import traceback tb = traceback.format_exc() LOG.error("Unexpected exception: %s" % tb) else: # no exception break # if it worked, exit the loop now rdict = r.json() if str(r) == "": swapi.LOG.debug("Response of GET request: %s" % str(r)) data = rdict.get("data",{}) info = data return True, r, info else: swapi.LOG.warning("Response of GET request: %s" % str(r)) message = rdict.get("message",None) info = message return False, r , info