Coverage for tubthumper/_interfaces.py: 100%
30 statements
« prev ^ index » next coverage.py v7.4.0, created at 2024-01-09 04:23 +0000
1"""Interfaces for tubthumper package"""
3import logging
4from typing import Callable
6from tubthumper import _types as tub_types
7from tubthumper._retry_factory import RetryConfig
8from tubthumper._retry_factory import retry_factory as _retry_factory
10__all__ = ["retry", "retry_decorator", "retry_factory"]
12RETRY_LIMIT_DEFAULT = float("inf")
13TIME_LIMIT_DEFAULT = float("inf")
14INIT_BACKOFF_DEFAULT = 1
15EXPONENTIAL_DEFAULT = 2
16JITTER_DEFAULT = True
17RERAISE_DEFAULT = False
18LOG_LEVEL_DEFAULT = logging.WARNING
19LOGGER_DEFAULT = logging.getLogger("tubthumper")
22def retry( # pylint: disable=too-many-arguments
23 func: Callable[tub_types.P, tub_types.T],
24 *,
25 exceptions: tub_types.Exceptions,
26 args: tub_types.Args = None,
27 kwargs: tub_types.Kwargs = None,
28 retry_limit: tub_types.RetryLimit = RETRY_LIMIT_DEFAULT,
29 time_limit: tub_types.Duration = TIME_LIMIT_DEFAULT,
30 init_backoff: tub_types.Duration = INIT_BACKOFF_DEFAULT,
31 exponential: tub_types.Exponential = EXPONENTIAL_DEFAULT,
32 jitter: tub_types.Jitter = JITTER_DEFAULT,
33 reraise: tub_types.Reraise = RERAISE_DEFAULT,
34 log_level: tub_types.LogLevel = LOG_LEVEL_DEFAULT,
35 logger: tub_types.Logger = LOGGER_DEFAULT,
36) -> tub_types.T:
37 r"""Call the provided callable with retry logic.
39 For usage examples, see `here <index.html#usage>`__.
41 Args:
42 func:
43 callable to be called
44 exceptions:
45 exceptions to be caught, resulting in a retry
46 args:
47 positional arguments for the callable
48 kwargs:
49 keyword arguments for the callable
50 retry_limit:
51 number of retries to perform before raising an exception,
52 e.g. ``retry_limit=1`` results in at most two calls
53 time_limit:
54 duration in seconds after which a retry attempt will
55 be prevented by raising an exception, i.e. not a timeout
56 stopping long running calls, but rather a mechanism to prevent
57 retry attempts after a certain duration
58 init_backoff:
59 duration in seconds to sleep before the first retry
60 exponential:
61 backoff duration between retries grows by this factor with each retry
62 jitter:
63 whether or not to "jitter" the backoff duration randomly
64 reraise:
65 whether or not to re-raise the caught exception instead of
66 a `RetryError` when a retry or time limit is reached
67 log_level:
68 level for logging caught exceptions, defaults to `logging.WARNING`
69 logger:
70 logger to log caught exceptions with
72 Raises:
73 RetryError:
74 Raised when a retry or time limit is reached, unless ``reraise=True``
76 Returns:
77 the returned object of the callable
78 """
79 if args is None:
80 args = ()
81 if kwargs is None:
82 kwargs = {}
83 retry_config = RetryConfig(
84 exceptions=exceptions,
85 retry_limit=retry_limit,
86 time_limit=time_limit,
87 init_backoff=init_backoff,
88 exponential=exponential,
89 jitter=jitter,
90 reraise=reraise,
91 log_level=log_level,
92 logger=logger,
93 )
94 retry_func = _retry_factory(func, retry_config)
95 return retry_func(*args, **kwargs)
98def retry_decorator( # pylint: disable=too-many-arguments
99 *,
100 exceptions: tub_types.Exceptions,
101 retry_limit: tub_types.RetryLimit = RETRY_LIMIT_DEFAULT,
102 time_limit: tub_types.Duration = TIME_LIMIT_DEFAULT,
103 init_backoff: tub_types.Duration = INIT_BACKOFF_DEFAULT,
104 exponential: tub_types.Exponential = EXPONENTIAL_DEFAULT,
105 jitter: tub_types.Jitter = JITTER_DEFAULT,
106 reraise: tub_types.Reraise = RERAISE_DEFAULT,
107 log_level: tub_types.LogLevel = LOG_LEVEL_DEFAULT,
108 logger: tub_types.Logger = LOGGER_DEFAULT,
109) -> Callable[[Callable[tub_types.P, tub_types.T]], Callable[tub_types.P, tub_types.T]]:
110 r"""Construct a decorator function for defining a function with built-in retry logic.
112 For usage examples, see `here <index.html#usage>`__.
114 Args:
115 exceptions:
116 exceptions to be caught, resulting in a retry
117 retry_limit:
118 number of retries to perform before raising an exception,
119 e.g. ``retry_limit=1`` results in at most two calls
120 time_limit:
121 duration in seconds after which a retry attempt will
122 be prevented by raising an exception, i.e. not a timeout
123 stopping long running calls, but rather a mechanism to prevent
124 retry attempts after a certain duration
125 init_backoff:
126 duration in seconds to sleep before the first retry
127 exponential:
128 backoff duration between retries grows by this factor with each retry
129 jitter:
130 whether or not to "jitter" the backoff duration randomly
131 reraise:
132 whether or not to re-raise the caught exception instead of
133 a `RetryError` when a retry or time limit is reached
134 log_level:
135 level for logging caught exceptions, defaults to `logging.WARNING`
136 logger:
137 logger to log caught exceptions with
139 Raises:
140 RetryError:
141 Raised when a retry or time limit is reached, unless ``reraise=True``
143 Returns:
144 a decorator function that, when used as such,
145 returns a function that looks like the callable it
146 decorates, but with configured retry logic
147 """
149 def decorator(
150 func: Callable[tub_types.P, tub_types.T]
151 ) -> Callable[tub_types.P, tub_types.T]:
152 retry_config = RetryConfig(
153 exceptions=exceptions,
154 retry_limit=retry_limit,
155 time_limit=time_limit,
156 init_backoff=init_backoff,
157 exponential=exponential,
158 jitter=jitter,
159 reraise=reraise,
160 log_level=log_level,
161 logger=logger,
162 )
163 return _retry_factory(func, retry_config)
165 return decorator
168def retry_factory( # pylint: disable=too-many-arguments
169 func: Callable[tub_types.P, tub_types.T],
170 *,
171 exceptions: tub_types.Exceptions,
172 retry_limit: tub_types.RetryLimit = RETRY_LIMIT_DEFAULT,
173 time_limit: tub_types.Duration = TIME_LIMIT_DEFAULT,
174 init_backoff: tub_types.Duration = INIT_BACKOFF_DEFAULT,
175 exponential: tub_types.Exponential = EXPONENTIAL_DEFAULT,
176 jitter: tub_types.Jitter = JITTER_DEFAULT,
177 reraise: tub_types.Reraise = RERAISE_DEFAULT,
178 log_level: tub_types.LogLevel = LOG_LEVEL_DEFAULT,
179 logger: tub_types.Logger = LOGGER_DEFAULT,
180) -> Callable[tub_types.P, tub_types.T]:
181 r"""Construct a function with built-in retry logic given a callable to retry.
183 For usage examples, see `here <index.html#usage>`__.
185 Args:
186 func:
187 callable to be called
188 exceptions:
189 exceptions to be caught, resulting in a retry
190 retry_limit:
191 number of retries to perform before raising an exception,
192 e.g. ``retry_limit=1`` results in at most two calls
193 time_limit:
194 duration in seconds after which a retry attempt will
195 be prevented by raising an exception, i.e. not a timeout
196 stopping long running calls, but rather a mechanism to prevent
197 retry attempts after a certain duration
198 init_backoff:
199 duration in seconds to sleep before the first retry
200 exponential:
201 backoff duration between retries grows by this factor with each retry
202 jitter:
203 whether or not to "jitter" the backoff duration randomly
204 reraise:
205 whether or not to re-raise the caught exception instead of
206 a `RetryError` when a retry or time limit is reached
207 log_level:
208 level for logging caught exceptions, defaults to `logging.WARNING`
209 logger:
210 logger to log caught exceptions with
212 Raises:
213 RetryError:
214 Raised when a retry or time limit is reached, unless ``reraise=True``
216 Returns:
217 a function that looks like the callable provided,
218 but with configured retry logic
219 """
220 retry_config = RetryConfig(
221 exceptions=exceptions,
222 retry_limit=retry_limit,
223 time_limit=time_limit,
224 init_backoff=init_backoff,
225 exponential=exponential,
226 jitter=jitter,
227 reraise=reraise,
228 log_level=log_level,
229 logger=logger,
230 )
231 return _retry_factory(func, retry_config)