qq_lib.core.retryer

Utility for retrying operations with configurable backoff.

This module provides the Retryer class, which executes a function repeatedly until it succeeds or a maximum number of attempts is reached. Failures are logged with timing information, and the final exception is re-raised with context when retries are exhausted.

View Source

 1# Released under MIT License.
 2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
 3
 4"""
 5Utility for retrying operations with configurable backoff.
 6
 7This module provides the `Retryer` class, which executes a function repeatedly
 8until it succeeds or a maximum number of attempts is reached. Failures are
 9logged with timing information, and the final exception is re-raised with
10context when retries are exhausted.
11"""
12
13from collections.abc import Callable
14from time import sleep
15from typing import Any
16
17from .error import QQError
18from .logger import get_logger
19
20logger = get_logger(__name__, show_time=True)
21
22
23class Retryer:
24    """
25    Retryer repeatedly executes a function until it succeeds or max attempts are reached.
26
27    Attributes:
28        func (Callable): The function or method to execute.
29        args (Tuple): Positional arguments to pass to the function.
30        kwargs (Dict): Keyword arguments to pass to the function.
31        max_tries (int): Maximum number of attempts.
32        wait_seconds (float): Time to wait between attempts.
33    """
34
35    def __init__(
36        self,
37        func: Callable,
38        *args: Any,
39        max_tries: int,
40        wait_seconds: float,
41        **kwargs: Any,
42    ):
43        self._func = func
44        self._args = args
45        self._kwargs = kwargs
46        self._max_tries = max_tries
47        self._wait_seconds = wait_seconds
48
49    def run(self) -> Any:
50        """
51        Execute the function repeatedly until it succeeds or max_tries is reached.
52
53        Any situation when the function does not raise an exception is considered a success.
54
55        Returns:
56            Any: The return value of the function if successful.
57
58        Raises:
59            Exception: The last exception raised if all attempts fail.
60        """
61        for attempt in range(1, self._max_tries + 1):
62            try:
63                return self._func(*self._args, **self._kwargs)
64            except Exception as e:
65                if attempt == self._max_tries:
66                    raise type(e)(
67                        f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempts exhausted."
68                    ) from e
69                logger.warning(
70                    f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempting again in {self._wait_seconds} seconds."
71                )
72                sleep(self._wait_seconds)
73
74        # should never get here
75        raise QQError(
76            "Execution got into an unexpected part of the Retryer.run method. This is a bug, please report it."
77        )

logger = <Logger qq_lib.core.retryer (INFO)>

class Retryer: View Source

24class Retryer:
25    """
26    Retryer repeatedly executes a function until it succeeds or max attempts are reached.
27
28    Attributes:
29        func (Callable): The function or method to execute.
30        args (Tuple): Positional arguments to pass to the function.
31        kwargs (Dict): Keyword arguments to pass to the function.
32        max_tries (int): Maximum number of attempts.
33        wait_seconds (float): Time to wait between attempts.
34    """
35
36    def __init__(
37        self,
38        func: Callable,
39        *args: Any,
40        max_tries: int,
41        wait_seconds: float,
42        **kwargs: Any,
43    ):
44        self._func = func
45        self._args = args
46        self._kwargs = kwargs
47        self._max_tries = max_tries
48        self._wait_seconds = wait_seconds
49
50    def run(self) -> Any:
51        """
52        Execute the function repeatedly until it succeeds or max_tries is reached.
53
54        Any situation when the function does not raise an exception is considered a success.
55
56        Returns:
57            Any: The return value of the function if successful.
58
59        Raises:
60            Exception: The last exception raised if all attempts fail.
61        """
62        for attempt in range(1, self._max_tries + 1):
63            try:
64                return self._func(*self._args, **self._kwargs)
65            except Exception as e:
66                if attempt == self._max_tries:
67                    raise type(e)(
68                        f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempts exhausted."
69                    ) from e
70                logger.warning(
71                    f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempting again in {self._wait_seconds} seconds."
72                )
73                sleep(self._wait_seconds)
74
75        # should never get here
76        raise QQError(
77            "Execution got into an unexpected part of the Retryer.run method. This is a bug, please report it."
78        )

Retryer repeatedly executes a function until it succeeds or max attempts are reached.

Attributes:

func (Callable): The function or method to execute.
args (Tuple): Positional arguments to pass to the function.
kwargs (Dict): Keyword arguments to pass to the function.
max_tries (int): Maximum number of attempts.
wait_seconds (float): Time to wait between attempts.

Retryer( func: Callable, *args: Any, max_tries: int, wait_seconds: float, **kwargs: Any) View Source

36    def __init__(
37        self,
38        func: Callable,
39        *args: Any,
40        max_tries: int,
41        wait_seconds: float,
42        **kwargs: Any,
43    ):
44        self._func = func
45        self._args = args
46        self._kwargs = kwargs
47        self._max_tries = max_tries
48        self._wait_seconds = wait_seconds

def run(self) -> Any: View Source

50    def run(self) -> Any:
51        """
52        Execute the function repeatedly until it succeeds or max_tries is reached.
53
54        Any situation when the function does not raise an exception is considered a success.
55
56        Returns:
57            Any: The return value of the function if successful.
58
59        Raises:
60            Exception: The last exception raised if all attempts fail.
61        """
62        for attempt in range(1, self._max_tries + 1):
63            try:
64                return self._func(*self._args, **self._kwargs)
65            except Exception as e:
66                if attempt == self._max_tries:
67                    raise type(e)(
68                        f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempts exhausted."
69                    ) from e
70                logger.warning(
71                    f"{e}\nThis was attempt {attempt} of {self._max_tries}. Attempting again in {self._wait_seconds} seconds."
72                )
73                sleep(self._wait_seconds)
74
75        # should never get here
76        raise QQError(
77            "Execution got into an unexpected part of the Retryer.run method. This is a bug, please report it."
78        )

Execute the function repeatedly until it succeeds or max_tries is reached.

Any situation when the function does not raise an exception is considered a success.

Returns:

Any: The return value of the function if successful.

Raises:

Exception: The last exception raised if all attempts fail.