Skip to content

Strategy

Retry strategies.

ExponentialBackoffStrategy

Bases: Strategy, Generic[StrategyValueT]

Exponential backoff strategy

Source code in pyretries/strategy.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
class ExponentialBackoffStrategy(Strategy, t.Generic[StrategyValueT]):
    """Exponential backoff strategy"""

    def __init__(self, max_attempts: int, base_delay: float, **kwargs) -> None:
        """
        Args:
            max_attempts: Number of maximum attempts
            base_delay: base delay in seconds for exponential backoff
            kwargs (dict): Passed to base strategy
        """
        self.base_delay = base_delay
        self.max_attempts = max_attempts
        self.current_attempt = 0
        self.delay = 0
        super().__init__(**kwargs)

    @property
    def should_stop(self) -> bool:
        """Check if strategy should apply.

        Return:
            Returns `True` if current attempt if greater or equal to `attempts`
        """
        if self.current_attempt >= self.max_attempts:
            return True
        return False

    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Return:
            Returns `True` if strategy should be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        if self.should_stop:
            raise RetryStrategyExausted from (
                value if isinstance(value, Exception) else None
            )

        self.current_attempt += 1
        self.delay = self.base_delay * 2**self.current_attempt + random.uniform(0, 1)

        if self.should_log:
            _logger.info(
                f"{self.__class__.__name__} {self.current_attempt}/{self.max_attempts} attempts."
                f" Sleeping for {self.delay:.2f}s"
            )

        time.sleep(self.delay)

should_stop: bool property

Check if strategy should apply.

Return

Returns True if current attempt if greater or equal to attempts

__init__(max_attempts, base_delay, **kwargs)

Parameters:

Name Type Description Default
max_attempts int

Number of maximum attempts

 required
base_delay float

base delay in seconds for exponential backoff

 required
kwargs dict

Passed to base strategy

 {}
Source code in pyretries/strategy.py 
265
266
267
268
269
270
271
272
273
274
275
276
def __init__(self, max_attempts: int, base_delay: float, **kwargs) -> None:
    """
    Args:
        max_attempts: Number of maximum attempts
        base_delay: base delay in seconds for exponential backoff
        kwargs (dict): Passed to base strategy
    """
    self.base_delay = base_delay
    self.max_attempts = max_attempts
    self.current_attempt = 0
    self.delay = 0
    super().__init__(**kwargs)

eval(value)

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required
Return

Returns True if strategy should be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Return:
        Returns `True` if strategy should be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    if self.should_stop:
        raise RetryStrategyExausted from (
            value if isinstance(value, Exception) else None
        )

    self.current_attempt += 1
    self.delay = self.base_delay * 2**self.current_attempt + random.uniform(0, 1)

    if self.should_log:
        _logger.info(
            f"{self.__class__.__name__} {self.current_attempt}/{self.max_attempts} attempts."
            f" Sleeping for {self.delay:.2f}s"
        )

    time.sleep(self.delay)

NoopStrategy

Bases: Strategy, Generic[StrategyValueT]

Do nothing strategy

Source code in pyretries/strategy.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class NoopStrategy(Strategy, t.Generic[StrategyValueT]):
    "Do nothing strategy"

    def __init__(self, attempts: int = 1, **kwargs) -> None:
        """
        Args:
            attempts: Number of maximum attempts
            kwargs (dict): Passed to base strategy
        """
        self.attempts = attempts
        self.current_attempt = 0
        super().__init__(**kwargs)

    @property
    def should_stop(self) -> bool:
        """Check if strategy should apply.

        Return:
            Returns `True` if current attempt if greater or equal to `attempts`
        """
        if self.current_attempt >= self.attempts:
            return True
        return False

    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Return:
            Returns `True` if strategy should be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        if self.should_stop:
            raise RetryStrategyExausted from (
                value if isinstance(value, Exception) else None
            )

        self.current_attempt += 1

should_stop: bool property

Check if strategy should apply.

Return

Returns True if current attempt if greater or equal to attempts

__init__(attempts=1, **kwargs)

Parameters:

Name Type Description Default
attempts int

Number of maximum attempts

 1
kwargs dict

Passed to base strategy

 {}
Source code in pyretries/strategy.py 
165
166
167
168
169
170
171
172
173
def __init__(self, attempts: int = 1, **kwargs) -> None:
    """
    Args:
        attempts: Number of maximum attempts
        kwargs (dict): Passed to base strategy
    """
    self.attempts = attempts
    self.current_attempt = 0
    super().__init__(**kwargs)

eval(value)

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required
Return

Returns True if strategy should be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Return:
        Returns `True` if strategy should be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    if self.should_stop:
        raise RetryStrategyExausted from (
            value if isinstance(value, Exception) else None
        )

    self.current_attempt += 1

SleepStrategy

Bases: Strategy, Generic[StrategyValueT]

Sleep strategy

Source code in pyretries/strategy.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class SleepStrategy(Strategy, t.Generic[StrategyValueT]):
    """Sleep strategy"""

    def __init__(self, seconds: float, attempts: int = 1, **kwargs):
        """
        Args:
            seconds: Amount of seconds to sleep
            attempts: Number of maximum attempts
            kwargs (dict): Passed to base strategy
        """
        self.seconds = seconds
        self.attempts = attempts
        self.current_attempt = 0
        super().__init__(**kwargs)

    @property
    def should_stop(self) -> bool:
        """Check if strategy should apply.

        Return:
            Returns `True` if current attempt if greater or equal to `attempts`
        """
        if self.current_attempt >= self.attempts:
            return True
        return False

    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Return:
            Returns `True` if strategy should be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        if self.should_stop:
            raise RetryStrategyExausted from (
                value if isinstance(value, Exception) else None
            )

        self.current_attempt += 1

        if self.should_log:
            _logger.info(
                f"{self.__class__.__name__} {self.current_attempt}/{self.attempts} attempts."
                f" Sleeping for {self.seconds}s"
            )
        time.sleep(self.seconds)

should_stop: bool property

Check if strategy should apply.

Return

Returns True if current attempt if greater or equal to attempts

__init__(seconds, attempts=1, **kwargs)

Parameters:

Name Type Description Default
seconds float

Amount of seconds to sleep

 required
attempts int

Number of maximum attempts

 1
kwargs dict

Passed to base strategy

 {}
Source code in pyretries/strategy.py 
111
112
113
114
115
116
117
118
119
120
121
def __init__(self, seconds: float, attempts: int = 1, **kwargs):
    """
    Args:
        seconds: Amount of seconds to sleep
        attempts: Number of maximum attempts
        kwargs (dict): Passed to base strategy
    """
    self.seconds = seconds
    self.attempts = attempts
    self.current_attempt = 0
    super().__init__(**kwargs)

eval(value)

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required
Return

Returns True if strategy should be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Return:
        Returns `True` if strategy should be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    if self.should_stop:
        raise RetryStrategyExausted from (
            value if isinstance(value, Exception) else None
        )

    self.current_attempt += 1

    if self.should_log:
        _logger.info(
            f"{self.__class__.__name__} {self.current_attempt}/{self.attempts} attempts."
            f" Sleeping for {self.seconds}s"
        )
    time.sleep(self.seconds)

StopAfterAttemptStrategy

Bases: Strategy, Generic[StrategyValueT]

Stop after attempting N times strategy

Source code in pyretries/strategy.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class StopAfterAttemptStrategy(Strategy, t.Generic[StrategyValueT]):
    """
    Stop after attempting N times strategy
    """

    def __init__(self, attempts: int, **kwargs) -> None:
        """
        Args:
            attempts: Number of attempts to run
            kwargs (dict): Passed to base strategy
        """
        self.attempts = attempts
        self.current_attempt = 0
        super().__init__(**kwargs)

    @property
    def should_stop(self) -> bool:
        """Check if strategy should apply.

        Return:
            Returns `True` if current attempt if greater or equal to `attempts`
        """
        if self.current_attempt >= self.attempts:
            return True
        return False

    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Return:
            Returns `True` if strategy should be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        if self.should_stop:
            raise RetryStrategyExausted from (
                value if isinstance(value, Exception) else None
            )

        if self.should_log:
            _logger.info(f"{self.__class__.__name__} is at {self.current_attempt=}")

        self.current_attempt += 1

should_stop: bool property

Check if strategy should apply.

Return

Returns True if current attempt if greater or equal to attempts

__init__(attempts, **kwargs)

Parameters:

Name Type Description Default
attempts int

Number of attempts to run

 required
kwargs dict

Passed to base strategy

 {}
Source code in pyretries/strategy.py 
63
64
65
66
67
68
69
70
71
def __init__(self, attempts: int, **kwargs) -> None:
    """
    Args:
        attempts: Number of attempts to run
        kwargs (dict): Passed to base strategy
    """
    self.attempts = attempts
    self.current_attempt = 0
    super().__init__(**kwargs)

eval(value)

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required
Return

Returns True if strategy should be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Return:
        Returns `True` if strategy should be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    if self.should_stop:
        raise RetryStrategyExausted from (
            value if isinstance(value, Exception) else None
        )

    if self.should_log:
        _logger.info(f"{self.__class__.__name__} is at {self.current_attempt=}")

    self.current_attempt += 1

StopWhenReturnValueStrategy

Bases: Strategy, Generic[StrategyValueT]

Stop when return value is X strategy

Source code in pyretries/strategy.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class StopWhenReturnValueStrategy(Strategy, t.Generic[StrategyValueT]):
    """Stop when return value is X strategy"""

    def __init__(
        self, expected: t.Any, max_attempts: int | None = None, **kwargs
    ) -> None:
        """
        Args:
            expected: Expected return value
            max_attempts: Number of maximum attempts. By default it runs forever
            kwargs (dict): Passed to base strategy
        """
        self.expected = expected
        self.max_attempts = max_attempts
        self.current_attempt = 0
        super().__init__(**kwargs)

    @property
    def should_stop(self) -> bool:
        """Check if strategy should apply.

        Return:
            Returns `True` if current attempt if greater or equal to `attempts`
        """
        if self.max_attempts is not None:
            if self.current_attempt >= self.max_attempts:
                return True
        return False

    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Return:
            Returns `True` if strategy should be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        if self.should_stop:
            raise RetryStrategyExausted from (
                value if isinstance(value, Exception) else None
            )

        if self.max_attempts is not None:
            if self.should_log:
                _logger.info(
                    f"{self.__class__.__name__} is at {self.current_attempt}/{self.max_attempts}."
                )
            self.current_attempt += 1

should_stop: bool property

Check if strategy should apply.

Return

Returns True if current attempt if greater or equal to attempts

__init__(expected, max_attempts=None, **kwargs)

Parameters:

Name Type Description Default
expected Any

Expected return value

 required
max_attempts int | None

Number of maximum attempts. By default it runs forever

 None
kwargs dict

Passed to base strategy

 {}
Source code in pyretries/strategy.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
def __init__(
    self, expected: t.Any, max_attempts: int | None = None, **kwargs
) -> None:
    """
    Args:
        expected: Expected return value
        max_attempts: Number of maximum attempts. By default it runs forever
        kwargs (dict): Passed to base strategy
    """
    self.expected = expected
    self.max_attempts = max_attempts
    self.current_attempt = 0
    super().__init__(**kwargs)

eval(value)

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required
Return

Returns True if strategy should be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Return:
        Returns `True` if strategy should be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    if self.should_stop:
        raise RetryStrategyExausted from (
            value if isinstance(value, Exception) else None
        )

    if self.max_attempts is not None:
        if self.should_log:
            _logger.info(
                f"{self.__class__.__name__} is at {self.current_attempt}/{self.max_attempts}."
            )
        self.current_attempt += 1

Strategy

Bases: ABC, Generic[StrategyValueT]

Base Strategy class

Source code in pyretries/strategy.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class Strategy(abc.ABC, t.Generic[StrategyValueT]):
    """Base Strategy class"""

    def __init__(self, should_log: bool = True) -> None:
        """
        Args:
            should_log: Specifies if strategy should log actions
        """
        self.should_log = should_log

    @abc.abstractproperty
    def should_stop(self) -> bool:
        """
        Checks if strategy should apply
        """
        raise NotImplementedError

    @abc.abstractmethod
    def eval(self, value: StrategyValueT | Exception | None):
        """
        Evaluates strategy.

        Args:
            value: `func` Returned value or exception.

        Returns:
            `True` if it should still be applied in the next iteration

        Raises:
            RetryStrategyExausted: Raised when strategy is exausted
        """
        raise NotImplementedError

__init__(should_log=True)

Parameters:

Name Type Description Default
should_log bool

Specifies if strategy should log actions

 True
Source code in pyretries/strategy.py 
27
28
29
30
31
32
def __init__(self, should_log: bool = True) -> None:
    """
    Args:
        should_log: Specifies if strategy should log actions
    """
    self.should_log = should_log

eval(value) abstractmethod

Evaluates strategy.

Parameters:

Name Type Description Default
value StrategyValueT | Exception | None

func Returned value or exception.

 required

Returns:

Type Description

True if it should still be applied in the next iteration

Raises:

Type Description
RetryStrategyExausted

Raised when strategy is exausted
Source code in pyretries/strategy.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@abc.abstractmethod
def eval(self, value: StrategyValueT | Exception | None):
    """
    Evaluates strategy.

    Args:
        value: `func` Returned value or exception.

    Returns:
        `True` if it should still be applied in the next iteration

    Raises:
        RetryStrategyExausted: Raised when strategy is exausted
    """
    raise NotImplementedError

should_stop()

Checks if strategy should apply

Source code in pyretries/strategy.py 
34
35
36
37
38
39
@abc.abstractproperty
def should_stop(self) -> bool:
    """
    Checks if strategy should apply
    """
    raise NotImplementedError