Skip to content

retry

frequenz.client.base.retry ¤

Implementations for retry strategies.

Attributes¤

frequenz.client.base.retry.DEFAULT_RETRY_INTERVAL module-attribute ¤

DEFAULT_RETRY_INTERVAL = 3.0

Default retry interval, in seconds.

frequenz.client.base.retry.DEFAULT_RETRY_JITTER module-attribute ¤

DEFAULT_RETRY_JITTER = 1.0

Default retry jitter, in seconds.

Classes¤

frequenz.client.base.retry.ExponentialBackoff ¤

Bases: Strategy

Provides methods for calculating the exponential interval between retries.

Source code in frequenz/client/base/retry.py
class ExponentialBackoff(Strategy):
    """Provides methods for calculating the exponential interval between retries."""

    DEFAULT_INTERVAL = DEFAULT_RETRY_INTERVAL
    """Default retry interval, in seconds."""

    DEFAULT_MAX_INTERVAL = 60.0
    """Default maximum retry interval, in seconds."""

    DEFAULT_MULTIPLIER = 2.0
    """Default multiplier for exponential increment."""

    def __init__(  # pylint: disable=too-many-arguments
        self,
        *,
        initial_interval: float = DEFAULT_INTERVAL,
        max_interval: float = DEFAULT_MAX_INTERVAL,
        multiplier: float = DEFAULT_MULTIPLIER,
        jitter: float = DEFAULT_RETRY_JITTER,
        limit: int | None = None,
    ) -> None:
        """Create a `ExponentialBackoff` instance.

        Args:
            initial_interval: time to wait for before the first retry, in
                seconds.
            max_interval: maximum interval, in seconds.
            multiplier: exponential increment for interval.
            jitter: a jitter to add to the retry interval.
            limit: max number of retries before giving up.  `None` means no
                limit, and `0` means no retry.
        """
        self._initial = initial_interval
        self._max = max_interval
        self._multiplier = multiplier
        self._jitter = jitter
        self._limit = limit

        self._count = 0

    def next_interval(self) -> float | None:
        """Return the time to wait before the next retry.

        Returns `None` if the retry limit has been reached, and no more retries
        are possible.

        Returns:
            Time until next retry when below retry limit, and None otherwise.
        """
        if self._limit is not None and self._count >= self._limit:
            return None
        self._count += 1
        exp_backoff_interval = self._initial * self._multiplier ** (self._count - 1)
        return min(exp_backoff_interval + random.uniform(0.0, self._jitter), self._max)
Attributes¤
DEFAULT_INTERVAL class-attribute instance-attribute ¤
DEFAULT_INTERVAL = DEFAULT_RETRY_INTERVAL

Default retry interval, in seconds.

DEFAULT_MAX_INTERVAL class-attribute instance-attribute ¤
DEFAULT_MAX_INTERVAL = 60.0

Default maximum retry interval, in seconds.

DEFAULT_MULTIPLIER class-attribute instance-attribute ¤
DEFAULT_MULTIPLIER = 2.0

Default multiplier for exponential increment.

Functions¤
__init__ ¤
__init__(
    *,
    initial_interval: float = DEFAULT_INTERVAL,
    max_interval: float = DEFAULT_MAX_INTERVAL,
    multiplier: float = DEFAULT_MULTIPLIER,
    jitter: float = DEFAULT_RETRY_JITTER,
    limit: int | None = None
) -> None

Create a ExponentialBackoff instance.

PARAMETER DESCRIPTION
initial_interval

time to wait for before the first retry, in seconds.

TYPE: float DEFAULT: DEFAULT_INTERVAL

max_interval

maximum interval, in seconds.

TYPE: float DEFAULT: DEFAULT_MAX_INTERVAL

multiplier

exponential increment for interval.

TYPE: float DEFAULT: DEFAULT_MULTIPLIER

jitter

a jitter to add to the retry interval.

TYPE: float DEFAULT: DEFAULT_RETRY_JITTER

limit

max number of retries before giving up. None means no limit, and 0 means no retry.

TYPE: int | None DEFAULT: None

Source code in frequenz/client/base/retry.py
def __init__(  # pylint: disable=too-many-arguments
    self,
    *,
    initial_interval: float = DEFAULT_INTERVAL,
    max_interval: float = DEFAULT_MAX_INTERVAL,
    multiplier: float = DEFAULT_MULTIPLIER,
    jitter: float = DEFAULT_RETRY_JITTER,
    limit: int | None = None,
) -> None:
    """Create a `ExponentialBackoff` instance.

    Args:
        initial_interval: time to wait for before the first retry, in
            seconds.
        max_interval: maximum interval, in seconds.
        multiplier: exponential increment for interval.
        jitter: a jitter to add to the retry interval.
        limit: max number of retries before giving up.  `None` means no
            limit, and `0` means no retry.
    """
    self._initial = initial_interval
    self._max = max_interval
    self._multiplier = multiplier
    self._jitter = jitter
    self._limit = limit

    self._count = 0
__iter__ ¤
__iter__() -> Iterator[float]

Return an iterator over the retry intervals.

YIELDS DESCRIPTION
float

Next retry interval in seconds.

Source code in frequenz/client/base/retry.py
def __iter__(self) -> Iterator[float]:
    """Return an iterator over the retry intervals.

    Yields:
        Next retry interval in seconds.
    """
    while True:
        interval = self.next_interval()
        if interval is None:
            break
        yield interval
copy ¤
copy() -> Self

Create a new instance of self.

RETURNS DESCRIPTION
Self

A deepcopy of self.

Source code in frequenz/client/base/retry.py
def copy(self) -> Self:
    """Create a new instance of `self`.

    Returns:
        A deepcopy of `self`.
    """
    ret = deepcopy(self)
    ret.reset()
    return ret
get_progress ¤
get_progress() -> str

Return a string denoting the retry progress.

RETURNS DESCRIPTION
str

String denoting retry progress in the form "(count/limit)"

Source code in frequenz/client/base/retry.py
def get_progress(self) -> str:
    """Return a string denoting the retry progress.

    Returns:
        String denoting retry progress in the form "(count/limit)"
    """
    if self._limit is None:
        return f"({self._count}/∞)"

    return f"({self._count}/{self._limit})"
next_interval ¤
next_interval() -> float | None

Return the time to wait before the next retry.

Returns None if the retry limit has been reached, and no more retries are possible.

RETURNS DESCRIPTION
float | None

Time until next retry when below retry limit, and None otherwise.

Source code in frequenz/client/base/retry.py
def next_interval(self) -> float | None:
    """Return the time to wait before the next retry.

    Returns `None` if the retry limit has been reached, and no more retries
    are possible.

    Returns:
        Time until next retry when below retry limit, and None otherwise.
    """
    if self._limit is not None and self._count >= self._limit:
        return None
    self._count += 1
    exp_backoff_interval = self._initial * self._multiplier ** (self._count - 1)
    return min(exp_backoff_interval + random.uniform(0.0, self._jitter), self._max)
reset ¤
reset() -> None

Reset the retry counter.

To be called as soon as a connection is successful.

Source code in frequenz/client/base/retry.py
def reset(self) -> None:
    """Reset the retry counter.

    To be called as soon as a connection is successful.
    """
    self._count = 0

frequenz.client.base.retry.LinearBackoff ¤

Bases: Strategy

Provides methods for calculating the interval between retries.

Source code in frequenz/client/base/retry.py
class LinearBackoff(Strategy):
    """Provides methods for calculating the interval between retries."""

    def __init__(
        self,
        *,
        interval: float = DEFAULT_RETRY_INTERVAL,
        jitter: float = DEFAULT_RETRY_JITTER,
        limit: int | None = None,
    ) -> None:
        """Create a `LinearBackoff` instance.

        Args:
            interval: time to wait for before the next retry, in seconds.
            jitter: a jitter to add to the retry interval.
            limit: max number of retries before giving up.  `None` means no
                limit, and `0` means no retry.
        """
        self._interval = interval
        self._jitter = jitter
        self._limit = limit

        self._count = 0

    def next_interval(self) -> float | None:
        """Return the time to wait before the next retry.

        Returns `None` if the retry limit has been reached, and no more retries
        are possible.

        Returns:
            Time until next retry when below retry limit, and None otherwise.
        """
        if self._limit is not None and self._count >= self._limit:
            return None
        self._count += 1
        return self._interval + random.uniform(0.0, self._jitter)
Functions¤
__init__ ¤
__init__(
    *,
    interval: float = DEFAULT_RETRY_INTERVAL,
    jitter: float = DEFAULT_RETRY_JITTER,
    limit: int | None = None
) -> None

Create a LinearBackoff instance.

PARAMETER DESCRIPTION
interval

time to wait for before the next retry, in seconds.

TYPE: float DEFAULT: DEFAULT_RETRY_INTERVAL

jitter

a jitter to add to the retry interval.

TYPE: float DEFAULT: DEFAULT_RETRY_JITTER

limit

max number of retries before giving up. None means no limit, and 0 means no retry.

TYPE: int | None DEFAULT: None

Source code in frequenz/client/base/retry.py
def __init__(
    self,
    *,
    interval: float = DEFAULT_RETRY_INTERVAL,
    jitter: float = DEFAULT_RETRY_JITTER,
    limit: int | None = None,
) -> None:
    """Create a `LinearBackoff` instance.

    Args:
        interval: time to wait for before the next retry, in seconds.
        jitter: a jitter to add to the retry interval.
        limit: max number of retries before giving up.  `None` means no
            limit, and `0` means no retry.
    """
    self._interval = interval
    self._jitter = jitter
    self._limit = limit

    self._count = 0
__iter__ ¤
__iter__() -> Iterator[float]

Return an iterator over the retry intervals.

YIELDS DESCRIPTION
float

Next retry interval in seconds.

Source code in frequenz/client/base/retry.py
def __iter__(self) -> Iterator[float]:
    """Return an iterator over the retry intervals.

    Yields:
        Next retry interval in seconds.
    """
    while True:
        interval = self.next_interval()
        if interval is None:
            break
        yield interval
copy ¤
copy() -> Self

Create a new instance of self.

RETURNS DESCRIPTION
Self

A deepcopy of self.

Source code in frequenz/client/base/retry.py
def copy(self) -> Self:
    """Create a new instance of `self`.

    Returns:
        A deepcopy of `self`.
    """
    ret = deepcopy(self)
    ret.reset()
    return ret
get_progress ¤
get_progress() -> str

Return a string denoting the retry progress.

RETURNS DESCRIPTION
str

String denoting retry progress in the form "(count/limit)"

Source code in frequenz/client/base/retry.py
def get_progress(self) -> str:
    """Return a string denoting the retry progress.

    Returns:
        String denoting retry progress in the form "(count/limit)"
    """
    if self._limit is None:
        return f"({self._count}/∞)"

    return f"({self._count}/{self._limit})"
next_interval ¤
next_interval() -> float | None

Return the time to wait before the next retry.

Returns None if the retry limit has been reached, and no more retries are possible.

RETURNS DESCRIPTION
float | None

Time until next retry when below retry limit, and None otherwise.

Source code in frequenz/client/base/retry.py
def next_interval(self) -> float | None:
    """Return the time to wait before the next retry.

    Returns `None` if the retry limit has been reached, and no more retries
    are possible.

    Returns:
        Time until next retry when below retry limit, and None otherwise.
    """
    if self._limit is not None and self._count >= self._limit:
        return None
    self._count += 1
    return self._interval + random.uniform(0.0, self._jitter)
reset ¤
reset() -> None

Reset the retry counter.

To be called as soon as a connection is successful.

Source code in frequenz/client/base/retry.py
def reset(self) -> None:
    """Reset the retry counter.

    To be called as soon as a connection is successful.
    """
    self._count = 0

frequenz.client.base.retry.Strategy ¤

Bases: ABC

Interface for implementing retry strategies.

Source code in frequenz/client/base/retry.py
class Strategy(ABC):
    """Interface for implementing retry strategies."""

    _limit: int | None
    _count: int

    @abstractmethod
    def next_interval(self) -> float | None:
        """Return the time to wait before the next retry.

        Returns `None` if the retry limit has been reached, and no more retries
        are possible.

        Returns:
            Time until next retry when below retry limit, and None otherwise.
        """

    def get_progress(self) -> str:
        """Return a string denoting the retry progress.

        Returns:
            String denoting retry progress in the form "(count/limit)"
        """
        if self._limit is None:
            return f"({self._count}/∞)"

        return f"({self._count}/{self._limit})"

    def reset(self) -> None:
        """Reset the retry counter.

        To be called as soon as a connection is successful.
        """
        self._count = 0

    def copy(self) -> Self:
        """Create a new instance of `self`.

        Returns:
            A deepcopy of `self`.
        """
        ret = deepcopy(self)
        ret.reset()
        return ret

    def __iter__(self) -> Iterator[float]:
        """Return an iterator over the retry intervals.

        Yields:
            Next retry interval in seconds.
        """
        while True:
            interval = self.next_interval()
            if interval is None:
                break
            yield interval
Functions¤
__iter__ ¤
__iter__() -> Iterator[float]

Return an iterator over the retry intervals.

YIELDS DESCRIPTION
float

Next retry interval in seconds.

Source code in frequenz/client/base/retry.py
def __iter__(self) -> Iterator[float]:
    """Return an iterator over the retry intervals.

    Yields:
        Next retry interval in seconds.
    """
    while True:
        interval = self.next_interval()
        if interval is None:
            break
        yield interval
copy ¤
copy() -> Self

Create a new instance of self.

RETURNS DESCRIPTION
Self

A deepcopy of self.

Source code in frequenz/client/base/retry.py
def copy(self) -> Self:
    """Create a new instance of `self`.

    Returns:
        A deepcopy of `self`.
    """
    ret = deepcopy(self)
    ret.reset()
    return ret
get_progress ¤
get_progress() -> str

Return a string denoting the retry progress.

RETURNS DESCRIPTION
str

String denoting retry progress in the form "(count/limit)"

Source code in frequenz/client/base/retry.py
def get_progress(self) -> str:
    """Return a string denoting the retry progress.

    Returns:
        String denoting retry progress in the form "(count/limit)"
    """
    if self._limit is None:
        return f"({self._count}/∞)"

    return f"({self._count}/{self._limit})"
next_interval abstractmethod ¤
next_interval() -> float | None

Return the time to wait before the next retry.

Returns None if the retry limit has been reached, and no more retries are possible.

RETURNS DESCRIPTION
float | None

Time until next retry when below retry limit, and None otherwise.

Source code in frequenz/client/base/retry.py
@abstractmethod
def next_interval(self) -> float | None:
    """Return the time to wait before the next retry.

    Returns `None` if the retry limit has been reached, and no more retries
    are possible.

    Returns:
        Time until next retry when below retry limit, and None otherwise.
    """
reset ¤
reset() -> None

Reset the retry counter.

To be called as soon as a connection is successful.

Source code in frequenz/client/base/retry.py
def reset(self) -> None:
    """Reset the retry counter.

    To be called as soon as a connection is successful.
    """
    self._count = 0