Created
June 19, 2022 13:26
-
-
Save thevickypedia/6d901bb123ed37996da49732fc67116d to your computer and use it in GitHub Desktop.
Retry a function for given number of times and handle known exceptions
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import functools | |
import logging | |
import random | |
from typing import Callable, Any | |
logging.root.setLevel(level=logging.DEBUG) | |
def retry(attempts: int = 3, exclude_exc=None) -> Callable: | |
"""Calls child func recursively. | |
Args: | |
attempts: Number of retry attempts. | |
exclude_exc: Exception that has to be logged and ignored. | |
Returns: | |
Callable: | |
Calls the decorator function. | |
""" | |
def decorator(func: Callable) -> Callable: | |
"""Calls the child func recursively. | |
Args: | |
func: Takes the function as an argument. Implemented as a decorator. | |
Returns: | |
Callable: | |
Calls the wrapper functon. | |
""" | |
@functools.wraps(func) | |
def wrapper(*args: Any, **kwargs: Any) -> Callable: | |
"""Executes the wrapped function in a loop for the number of attempts mentioned. | |
Args: | |
*args: Arguments. | |
**kwargs: Keyword arguments. | |
Returns: | |
Callable: | |
Return value of the function implemented. | |
Raises: | |
Raises the exception as received beyond the given number of attempts. | |
""" | |
return_exc = None | |
for i in range(attempts): | |
try: | |
return_val = func(*args, **kwargs) | |
logging.info(msg=f"Succeeded in attempt: {i}") | |
return return_val | |
except exclude_exc or KeyboardInterrupt as error: | |
logging.error(msg=error) | |
except Exception as e: | |
return_exc = e | |
logging.error(msg=f"Exceeded retry count::{attempts}") | |
if return_exc: | |
raise return_exc | |
return wrapper | |
return decorator | |
@retry(attempts=10) | |
def function_to_execute(): | |
randint = random.randint(1, 100) | |
if randint == 2: | |
return randint | |
raise IndexError | |
if __name__ == '__main__': | |
print(function_to_execute()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment