Coverage for src/toolbox_python/retry.py: 100%
58 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-02-24 10:34 +0000
1# ============================================================================ #
2# #
3# Title : Retry #
4# Purpose : Automatically retry a given function when a specific #
5# `Exception` is thrown. #
6# #
7# ============================================================================ #
10# ---------------------------------------------------------------------------- #
11# #
12# Overview ####
13# #
14# ---------------------------------------------------------------------------- #
17# ---------------------------------------------------------------------------- #
18# Description ####
19# ---------------------------------------------------------------------------- #
22"""
23!!! note "Summary"
24 The `retry` module is for enabling automatic retrying of a given function when a specific `Exception` is thrown.
25"""
28# ---------------------------------------------------------------------------- #
29# #
30# Setup ####
31# #
32# ---------------------------------------------------------------------------- #
35# ---------------------------------------------------------------------------- #
36# Imports ####
37# ---------------------------------------------------------------------------- #
40# ## Python StdLib Imports ----
41import inspect
42import logging
43from builtins import Exception
44from functools import wraps
45from logging import Logger
46from time import sleep
47from types import ModuleType
48from typing import Any, Callable, Literal, Optional, Union
50# ## Python Third Party Imports ----
51from typeguard import typechecked
53# ## Local First Party Imports ----
54from toolbox_python.classes import get_full_class_name
55from toolbox_python.collection_types import str_list
56from toolbox_python.output import print_or_log_output
59# ---------------------------------------------------------------------------- #
60# Exports ####
61# ---------------------------------------------------------------------------- #
64__all__: str_list = ["retry"]
67# ---------------------------------------------------------------------------- #
68# Types ####
69# ---------------------------------------------------------------------------- #
72_exceptions = Union[
73 type[Exception],
74 list[type[Exception]],
75 tuple[type[Exception], ...],
76]
77"""
78!!! note "Summary"
79 This
80"""
83# ---------------------------------------------------------------------------- #
84# #
85# Classes ####
86# #
87# ---------------------------------------------------------------------------- #
90class Retry:
91 pass
94# ---------------------------------------------------------------------------- #
95# #
96# Functions ####
97# #
98# ---------------------------------------------------------------------------- #
101@typechecked
102def retry(
103 exceptions: _exceptions = Exception,
104 tries: int = 1,
105 delay: int = 0,
106 print_or_log: Optional[Literal["print", "log"]] = "print",
107) -> Optional[Any]:
108 """
109 !!! note "Summary"
110 Retry a given function a number of times. Catching any known exceptions when they are given. And retrurning any output to either a terminal or a log file.
112 !!! deprecation "Deprecated"
113 This function is deprecated. Please use the [`retry()`][func] decorator from the [`stamina`][docs] package instead.<br>
114 For more info, see: [Docs][docs], [GitHub][github], [PyPi][pypi].
115 [func]: https://stamina.hynek.me/en/stable/api.html#stamina.retry
116 [docs]: https://stamina.hynek.me/en/stable/index.html
117 [github]: https://github.com/hynek/stamina/
118 [pypi]: https://pypi.org/project/stamina/
120 ???+ abstract "Details"
121 This function should always be implemented as a decorator.<br>
122 It is written based on the premise that a certain process may fail and return a given message, but that is known and expected, and you just want to wait a second or so then retry again.<br>
123 Typically, this is seen in async processes, or when writing data to a `delta` table when there may be concurrent read/writes occurring at the same time. In these instances, you will know the error message and can re-try again a certain number of times.
125 Params:
126 exceptions (_exceptions, optional):
127 A given single or collection of expected exceptions for which to catch and retry for.<br>
128 Defaults to `#!py Exception`.
129 tries (int, optional):
130 The number of retries to attempt. If the underlying process is still failing after this number of attempts, then throw a hard error and alert the user.<br>
131 Defaults to `#!py 1`.
132 delay (int, optional):
133 The number of seconds to delay between each retry.<br>
134 Defaults to `#!py 0`.
135 print_or_log (Optional[Literal["print", "log"]], optional):
136 Whether or not the messages should be written to the terminal in a `#!py print()` statement, or to a log file in a `#!py log()` statement.<br>
137 Defaults to `#!py "print"`.
139 Raises:
140 TypeError: If any of the inputs parsed to the parameters of this function are not the correct type. Uses the [`@typeguard.typechecked`](https://typeguard.readthedocs.io/en/stable/api.html#typeguard.typechecked) decorator.
141 ValueError: If either `tries` or `delay` are less than `#!py 0`
142 RuntimeError: If _either_ an unexpected `#!py Exception` was thrown, which was not declared in the `exceptions` collection, _or_ if the `func` was still not able to be executed after `tries` number of iterations.
144 Returns:
145 result (Optional[Any]):
146 The result from the underlying function, if any.
148 ???+ example "Examples"
150 ```{.py .python linenums="1" title="Imports"}
151 >>> from toolbox_python.retry import retry
152 ```
154 ```{.py .python linenums="1" title="Example 1: No error"}
155 >>> @retry(tries=5, delay=1, print_or_log="print")
156 >>> def simple_func(var1:str="this")->str:
157 ... return var1
158 >>> simple_func()
159 ```
160 <div class="result" markdown>
161 ```{.sh .shell title="Terminal"}
162 # No error
163 ```
164 </div>
166 ```{.py .python linenums="1" title="Example 2: Expected error"}
167 >>> @retry(exceptions=TypeError, tries=5, delay=1, print_or_log="print")
168 >>> def failing_func(var1:str="that")->None:
169 ... raise ValueError("Incorrect value")
170 >>> failing_func()
171 ```
172 <div class="result" markdown>
173 ```{.sh .shell title="Terminal"}
174 Caught an expected error at iteration 1: `ValueError`. Retrying in 1 seconds...
175 Caught an expected error at iteration 2: `ValueError`. Retrying in 1 seconds...
176 Caught an expected error at iteration 3: `ValueError`. Retrying in 1 seconds...
177 Caught an expected error at iteration 4: `ValueError`. Retrying in 1 seconds...
178 Caught an expected error at iteration 5: `ValueError`. Retrying in 1 seconds...
179 RuntimeError: Still could not write after 5 iterations. Please check.
180 ```
181 </div>
183 ??? success "Credit"
184 Inspiration from:
186 - https://pypi.org/project/retry/
187 - https://stackoverflow.com/questions/21786382/pythonic-way-of-retry-running-a-function#answer-21788594
188 """
189 for param in ["tries", "delay"]:
190 if not eval(param) >= 0:
191 raise ValueError(
192 f"Invalid value for parameter `{param}`: {eval(param)}\n"
193 f"Must be a positive integer."
194 )
195 if print_or_log == "log":
196 stk: inspect.FrameInfo = inspect.stack()[2]
197 mod: Union[ModuleType, None] = inspect.getmodule(stk[0])
198 if mod is not None:
199 log: Optional[Logger] = logging.getLogger(mod.__name__)
200 else:
201 log = None
203 def decorator(func: Callable):
204 @wraps(func)
205 def result(*args, **kwargs):
206 for i in range(1, tries + 1):
207 try:
208 results = func(*args, **kwargs)
209 except exceptions as e:
210 # Catch raw exceptions as defined in the `exceptions` parameter.
211 message = (
212 f"Caught an expected error at iteration {i}: "
213 f"`{get_full_class_name(e)}`. "
214 f"Retrying in {delay} seconds..."
215 )
216 print_or_log_output(
217 message=message,
218 print_or_log=print_or_log,
219 log=log,
220 log_level="warning",
221 )
222 sleep(delay)
223 except Exception as exc:
224 """
225 Catch unknown exception, however still need to check whether the name of any of the exceptions defined in `exceptions` are somehow listed in the text output of the caught exception.
226 The cause here is shown in the below chunk. You see here that it throws a 'Py4JJavaError', which was not listed in the `exceptions` parameter, yet within the text output, it showed the 'ConcurrentDeleteReadException' which _was_ listed in the `exceptions` parameter. Therefore, in this instance, we still want to sleep and retry
228 >>> Caught an unexpected error at iteration 1: `py4j.protocol.Py4JJavaError`.
229 >>> Time for fct_Receipt: 27secs
230 >>> java.util.concurrent.ExecutionException: io.delta.exceptions.
231 ... ConcurrentDeleteReadException: This transaction attempted to read one or more files that were deleted (for example part-00001-563449ea-73e4-4d7d-8ba8-53fee1f8a5ff.c000.snappy.parquet in the root of the table) by a concurrent update. Please try the operation again.
232 """
233 excs = (
234 [exceptions]
235 if not isinstance(exceptions, (list, tuple))
236 else exceptions
237 )
238 exc_names = [exc.__name__ for exc in excs]
239 if any(name in f"{exc}" for name in exc_names):
240 caught_error = [name for name in exc_names if name in f"{exc}"]
241 message = (
242 f"Caught an unexpected, known error at iteration {i}: "
243 f"`{get_full_class_name(exc)}`.\n"
244 f"Who's message contains reference to underlying exception(s): {caught_error}.\n"
245 f"Retrying in {delay} seconds..."
246 )
247 print_or_log_output(
248 message=message,
249 print_or_log=print_or_log,
250 log=log,
251 log_level="warning",
252 )
253 sleep(delay)
254 else:
255 message = (
256 f"Caught an unexpected error at iteration {i}: "
257 f"`{get_full_class_name(exc)}`."
258 )
259 print_or_log_output(
260 message=message,
261 print_or_log=print_or_log,
262 log=log,
263 log_level="error",
264 )
265 raise RuntimeError(message) from exc
266 else:
267 message = f"Successfully executed at iteration {i}."
268 print_or_log_output(
269 message=message,
270 print_or_log=print_or_log,
271 log=log,
272 log_level="info",
273 )
274 return results
275 message = f"Still could not write after {tries} iterations. Please check."
276 print_or_log_output(
277 message=message,
278 print_or_log=print_or_log,
279 log=log,
280 log_level="error",
281 )
282 raise RuntimeError(message)
284 return result
286 return decorator