Overengineered solution to retrying and exceptions in python
Goal: retry running function X times max Scenario: networking-ish issues
Solution: I came up with the thing below. It gets an optional list of acceptable exception types, and retries N times every time it gets one of them. As soon as it gets an unacceptable exception it passes it further. As soon as the function runs successfully it returns the function’s return value.
Can repeat infinite times and can consider all exceptions acceptable if both params are given empty or None.
from urllib3.exceptions import ProtocolError
from functools import partial
from itertools import count
from typing import Optional
def _try_n_times(fn, n_times: Optional[int]=3, acceptable_exceptions: Optional[tuple] =(ProtocolError, )):
""" Try function X times before giving up.
Concept:
- retry N times if fn fails with an acceptable exception
- raise immediately any exceptions not inside acceptable_exceptions
- if n_times is falsey will retry infinite times
- if acceptable_exceptions is falsey, all exceptions are acceptable
Returns:
- after n<n_times retries the return value of the first successdful run of fn
Raises:
- first unacceptable exceptions if acceptable_exceptions is not empty
- last exception raised by fn after too many retries
Args:
fn: callable to run
n_times: how many times, 0 means infinite
acceptable_exceptions: iterable of exceptions classes after which retry
empty/None means all exceptions are OK
TODO: if this works, integrate into load image/json as well (or increase
the number of retries organically) for e.g. NameResolutionErrors
and similar networking/connection issues
"""
last_exc = None
for time in range(n_times) if n_times else count(0):
try:
# Try running the function and save output
# break if it worked
if time>0:
logger.debug(f"Running fn {time=}")
res = fn()
break
except Exception as e:
# If there's an exception, raise bad ones otherwise continue the loop
if acceptable_exceptions and e.__class__ not in acceptable_exceptions:
logger.error(f"Caught {e} not in {acceptable_exceptions=}, so raising")
raise
logger.debug(f"Caught acceptable {e} our {time}'th time, continuing")
last_exc = e
continue
else:
# If loop went through without a single break it means fn always failed
# we raise the last exception
logger.error(f"Went through {time} acceptable exceptions, all failed, last exception was {last_exc}")
raise last_exc
# Return whatever fn returned on its first successful run
return res
The main bit here was that I didn’t want to use any magic values that might conflict with whatever the function returns (if I get a None/False how can I know it wasn’t the function without ugly complex magic values?)
The main insight here is the else clause w/ break.
fn is run as fn() and partial is a good way to generate them
EDIT: (ty CH) you can also just declare a function, lol