gh-126927: Add functools.retry decorator function

Doc/whatsnew/3.14.rst

@@ -324,6 +324,9 @@ functools
   as a keyword argument.
   (Contributed by Sayandip Dutta in :gh:`125916`.)
 
+* Add function decorator ``functools.retry`` that allows configurable
+  retries for functions that raise an exception.  It can be used with or
+  without keyword arguments.
 
 getopt
 ------

Lib/functools.py

@@ -12,7 +12,7 @@
 __all__ = ['update_wrapper', 'wraps', 'WRAPPER_ASSIGNMENTS', 'WRAPPER_UPDATES',
            'total_ordering', 'cache', 'cmp_to_key', 'lru_cache', 'reduce',
            'partial', 'partialmethod', 'singledispatch', 'singledispatchmethod',
-           'cached_property', 'Placeholder']
+           'cached_property', 'Placeholder', 'retry']
 
 from abc import get_cache_token
 from collections import namedtuple
@@ -21,6 +21,7 @@
 from reprlib import recursive_repr
 from types import GenericAlias, MethodType, MappingProxyType, UnionType
 from _thread import RLock
+from time import sleep
 
 ################################################################################
 ### update_wrapper() and wraps() decorator
@@ -1121,3 +1122,50 @@ def __get__(self, instance, owner=None):
         return val
 
     __class_getitem__ = classmethod(GenericAlias)
+
+
+def retry(
+    _kwargs=None, *, retry_attempts=3, interval_seconds=0.1, backoff_type="linear"
+):
+    """
+    This function is intended to be used as a decorator and will retry
+    the function that it decorates if an excpetion is raised in that
+    function.  Several aspects of the retries can be configured with
+    keyword arguments.  Also, no keyword arguments can be used to retry
+    with the default values.
+
+    NOTE: if using backoff_type='exponential', ensure that
+    interval_seconds > 1, otherise the subsequent retries will be
+    shorter.
+    """
+
+    def _retry(user_function):
+        @wraps(user_function)
+        def _retry_wrapper_user_function(*args, **kwargs):
+            for attempt_number in range(retry_attempts + 1):
+                try:
+                    return_value = user_function(*args, **kwargs)
+                    break
+                except Exception as e:
+                    if attempt_number < retry_attempts:
+                        if backoff_type == "exponential":
+                            # If user inputs interval_seconds < 1 with exponential backoff, retries will get shorter
+                            sleep(interval_seconds * (attempt_number + 1) ** 2)
+                        elif backoff_type == "linear":
+                            sleep(interval_seconds)
+                    else:
+                        # Retry attempts reached, raise the last exception
+                        raise e
+            return return_value
+
+        return _retry_wrapper_user_function
+
+    if backoff_type != "linear" and backoff_type != "exponential":
+        raise TypeError(
+            "Keyword argument backoff_type must be 'exponential' or 'linear'."
+        )
+
+    if _kwargs is None:
+        return _retry
+    else:
+        return _retry(_kwargs)

Lib/test/test_functools.py

@@ -3381,5 +3381,69 @@ def prop(self):
         self.assertEqual(t.prop, 1)
 
 
+class TestRetry(unittest.TestCase):
+
+    def test_function_fail(self):
+
+        @functools.retry
+        def fail_function1():
+            raise ValueError
+
+        with self.assertRaises(ValueError):
+            fail_function1()
+
+        @functools.retry(interval_seconds=.1, retry_attempts=1, backoff_type='exponential')
+        def fail_function2():
+            raise ValueError
+
+        with self.assertRaises(ValueError):
+            fail_function2()
+
+    def test_function_success(self):
+
+        @functools.retry
+        def success_function(a, b, c='test_value'):
+            return (a, b, c)
+
+        value = success_function('a', 'b')
+
+        self.assertEqual(value, ('a', 'b', 'test_value'))
+
+        class TestObject:
+            def __init__(self, call_count):
+                self.call_count = call_count
+        test_object = TestObject(0)
+
+        @functools.retry(interval_seconds=.01, retry_attempts=3, backoff_type='exponential')
+        def success_function_after_3_failures(test_object):
+            test_object.call_count += 1
+            if test_object.call_count > 3:
+                return True
+            raise ValueError('Some error message!')
+
+        value = success_function_after_3_failures(test_object)
+        self.assertTrue(value)
+
+    def test_backoff_type(self):
+        @functools.retry(backoff_type='exponential')
+        def user_function1():
+            return True
+        value1 = user_function1()
+        self.assertTrue(value1)
+
+        @functools.retry(backoff_type='linear')
+        def user_function2():
+            return True
+        value2 = user_function2()
+        self.assertTrue(value2)
+
+        with self.assertRaises(TypeError):
+            @functools.retry(backoff_type='incorrect_value')
+            def user_function3():
+                return True
+
+            user_function3()
+
+
 if __name__ == '__main__':
     unittest.main()

Misc/NEWS.d/next/Library/2024-11-18-03-26-52.gh-issue-126927.K-gKnU.rst

@@ -0,0 +1,4 @@
+Add ``functools.retry`` function to be used as a decorator.  The
+decorator gives retries to a function that raises exeptions.  It can be
+used with default options or have options specified with keyword
+arguments.