Skip to content

Instantly share code, notes, and snippets.

@dutc
Last active February 21, 2023 19:32
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dutc/bb804dc003c87cd2348a4ea64220fbef to your computer and use it in GitHub Desktop.
Save dutc/bb804dc003c87cd2348a4ea64220fbef to your computer and use it in GitHub Desktop.
Legitimately Bad Idea (`retry` decorator)
from functools import wraps
from itertools import islice, tee, zip_longest, chain, product
from collections import deque
from pandas import DataFrame
nwise = lambda g, *, n=2: zip(*(islice(g, i, None) for i, g in enumerate(tee(g, n))))
nwise_longest = lambda g, *, n=2, fv=object(): zip_longest(*(islice(g, i, None) for i, g in enumerate(tee(g, n))), fillvalue=fv)
first = lambda g, *, n=1: zip(chain(repeat(True, n), repeat(False)), g)
last = lambda g, *, m=1, s=object(): ((y[-1] is s, x) for x, *y in nwise_longest(g, n=m+1, fv=s))
def retry(*exc_types, tries=2): # XXX: a legitimately bad idea
''' allows you to dynamically retry a function and ignore certain exceptions '''
exc_types = exc_types if exc_types else (Exception,)
def dec(f):
@wraps(f)
def inner(*args, **kwargs):
for is_last, _ in last(range(tries)):
try:
return f(*args, **kwargs)
except exc_types as e:
if is_last: raise
continue
return inner
return dec
@retry(AssertionError)
def f(xs):
assert xs.popleft()
if __name__ == '__main__':
def res_or_err(f, *args, **kwargs):
try: return f(*args, **kwargs)
except Exception as e: return e
results = DataFrame.from_records([
(try1, try2, not isinstance(res_or_err(f, deque([try1, try2])), Exception))
for try1, try2 in product([True, False], repeat=2)
], columns='try1 try2 succeeds'.split()).set_index(['try1', 'try2']).squeeze(axis='columns')
print(results)
from collections import namedtuple
from pandas import DataFrame
from itertools import product
from collections import deque
# XXX: probably overkill but not an inherently bad idea
class Result(namedtuple('Result', 'res err')):
success = property(lambda self: self.err is None)
failure = property(lambda self: self.err is not None)
from_result = classmethod(lambda cls, res: cls(res=res, err=None))
from_error = classmethod(lambda cls, err: cls(res=None, err=err))
@classmethod
def from_call(cls, f, *args, **kwargs):
try:
return cls.from_result(f(*args, **kwargs))
except BaseException as err:
return cls.from_error(err)
def f(xs):
assert xs.popleft()
if __name__ == '__main__':
results = DataFrame.from_records([
(try1, try2, next((r for _ in range(2) if (r := Result.from_call(f, xs).success)), False))
for try1, try2 in product([True, False], repeat=2)
for xs in (deque([try1, try2]),)
], columns='try1 try2 succeeds'.split()).set_index(['try1', 'try2']).squeeze(axis='columns')
print(results)
@kmader
Copy link

kmader commented Feb 21, 2023

Lol, literally one of the first wrappers I wrote after becoming a professional software engineer 🤷

@dutc
Copy link
Author

dutc commented Feb 21, 2023

Here are some things which the second approach allows (with no additional library code!) that the first does not (without significant rewriting):

  • what if we want to retry but know how many times we retried?
  • what if we don't want to retry a fixed number of times, but until a certain time limit has been hit? (i.e., use an arbitrary external condition for determining whether to retry)
  • what if we want to slightly change what we do on the retry?
  • what if we want to put a delay between each retry?
  • what if we want to look at the Exception payload and decide to retry?
  • what if we want to want to retry for reasons other than an Exception (e.g., for certain return values)?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment