Skip to content

Instantly share code, notes, and snippets.

@SteadBytes
Last active April 10, 2021 00:25
Show Gist options
  • Save SteadBytes/ed517501b25b47fcc4a136b6578972b4 to your computer and use it in GitHub Desktop.
Save SteadBytes/ed517501b25b47fcc4a136b6578972b4 to your computer and use it in GitHub Desktop.
import time
from functools import wraps
class RetryExhaustedError(Exception):
pass
def retry(*exceptions, retries=5, cooldown=1, verbose=True):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retry_count = 0
while True:
try:
result = func(*args, **kwargs)
except exceptions as err:
retry_count += 1
msg = (
f"Exception during {func} execution."
f"{retry_count} of {retries} retries attempted"
)
if retry_count > retries:
verbose and logger.exception(msg)
raise RetryExhaustedError(
func.__qualname__, args, kwargs
) from err
else:
verbose and logger.warning(msg)
if cooldown:
time.sleep(cooldown)
else:
return result
return wrapper
return decorator
@retry(requests.HTTPError)
def fetch_data():
r = requests.get("https://httpbin.org/status/200%2C400%2404%2500")
r.raise_for_status()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment