Timers

A Timer found in the timer module simply stores a float or int value and increments it based on some mathematical formula. Sleep operations can then be called on the Timer to pause operation for the number of seconds specified by the value of the Timer.

The main motivation for this is to provide a flexible framework for managing retries and backoff for the RequestHandler.

See also

See Managing retries and backoff time for more info in how a Timer can be used with RequestHandler to handle retries and backoff.

This page gives an overview of some of the Timer implementations provided in this module, with info on how to implement your own.

Basic usage

from aiorequestful.timer import Timer

timer = Timer()

timer.increase()  # increase the value
timer.wait()  # wait for the number of seconds specified by the timer's current value synchronously


async def wait() -> None:
    await timer  # as above, but wait asynchronously

value_int = int(timer)  # get the current value as an int
value_float = float(timer)  # get the current value as a float

timer.reset()  # reset the timer back to its initial settings

See also

This module implements a few basic timing formulae as shown below, though you may wish to extend this functionality.

CountTimer

Provides an abstract implementation for managing a number of Timer value increases specified by a given count value.

StepCountTimer

Increases the timer value by a given step amount a distinct number of times.

from aiorequestful.timer import StepCountTimer

timer = StepCountTimer(initial=0, count=3, step=0.2)  # value = 0
timer.increase()  # value = 0.2
timer.increase()  # value = 0.4
timer.increase()  # value = 0.6
timer.increase()  # value = 0.6 (max count of 3 reached)

print(float(timer))

GeometricCountTimer

Increases the timer value by multiplying the current value by a given factor a distinct number of times.

from aiorequestful.timer import GeometricCountTimer

timer = GeometricCountTimer(initial=2, count=3, factor=2)  # value = 2
timer.increase()  # value = 4
timer.increase()  # value = 8
timer.increase()  # value = 16
timer.increase()  # value = 16 (max count of 3 reached)

print(float(timer))

PowerCountTimer

Increases the timer value by raising the current value to a given exponent a distinct number of times.

from aiorequestful.timer import PowerCountTimer

timer = PowerCountTimer(initial=2, count=3, exponent=2)  # value = 2
timer.increase()  # value = 4
timer.increase()  # value = 16
timer.increase()  # value = 256
timer.increase()  # value = 256 (max count of 3 reached)

print(float(timer))

CeilingTimer

Provides an abstract implementation for managing Timer value increases up to a specified final value.

StepCeilingTimer

Increases the timer value by a given step amount until a maximum value is reached.

from aiorequestful.timer import StepCeilingTimer

timer = StepCeilingTimer(initial=0, final=0.5, step=0.2)  # value = 0
timer.increase()  # value = 0.2
timer.increase()  # value = 0.4
timer.increase()  # value = 0.5 (max value reached)
timer.increase()  # value = 0.5

print(float(timer))

GeometricCeilingTimer

Increases the timer value by multiplying the current value by a given factor until a maximum value is reached.

from aiorequestful.timer import GeometricCeilingTimer

timer = GeometricCeilingTimer(initial=2, final=10, factor=2)  # value = 2
timer.increase()  # value = 4
timer.increase()  # value = 8
timer.increase()  # value = 10 (max value reached)
timer.increase()  # value = 10

print(float(timer))

PowerCeilingTimer

Increases the timer value by raising the current value to a given exponent until a maximum value is reached.

from aiorequestful.timer import PowerCeilingTimer

timer = PowerCeilingTimer(initial=2, final=60, exponent=2)  # value = 2
timer.increase()  # value = 4
timer.increase()  # value = 16
timer.increase()  # value = 60 (max value reached)
timer.increase()  # value = 60

print(float(timer))

Writing a Timer

To implement a Timer, you will need to implement the abstract methods as shown below.

@functools.total_ordering
class Timer(SupportsInt, SupportsFloat, ABC):
    """
    Base interface for all timers.

    :param initial: The starting value to use.
    """

    __slots__ = ("_initial", "_value", "_counter")

    @property
    def initial(self) -> Number:
        """The initial starting timer value in seconds."""
        return self._initial

    @property
    @abstractmethod
    def final(self) -> Number | None:
        """The maximum possible timer value in seconds."""
        raise NotImplementedError

    @property
    @abstractmethod
    def total(self) -> Number | None:
        """The sum of all possible timer values in seconds."""
        raise NotImplementedError

    @property
    @abstractmethod
    def total_remaining(self) -> Number | None:
        """The sum of all possible remaining timer values in seconds not including the current value."""
        raise NotImplementedError

    @property
    @abstractmethod
    def count(self) -> int | None:
        """The total amount of times this timer can be increased."""
        raise NotImplementedError

    @property
    def counter(self) -> int | None:
        """The number of times this timer has been increased."""
        return self._counter

    @property
    def count_remaining(self) -> int | None:
        """The remaining number of times this timer can be increased."""
        if self.count is None:
            return None
        return self.count - self.counter

    @property
    @abstractmethod
    def can_increase(self) -> bool:
        """Check whether this timer can be increased"""
        raise NotImplementedError

    def __init__(self, initial: Number = 0):
        self._initial = initial
        self._value = initial
        self._counter = 0

    def __int__(self):
        return int(self._value)

    def __float__(self):
        return float(self._value)

    def __eq__(self, other: Number):
        return self._value == other

    def __lt__(self, other: Number):
        return self._value < other

    def __round__(self, n: int = None) -> float:
        return round(float(self._value), n)

    def __await__(self) -> Generator[None, None, None]:
        """Asynchronously sleep for the current time set for this timer."""
        return asyncio.sleep(self._value).__await__()

    def __call__(self) -> None:
        """Sleep for the current time set for this timer."""
        return self.wait()

    def __deepcopy__(self, memo: dict):
        cls = self.__class__
        obj = cls.__new__(cls)

        memo[id(self)] = obj
        slots = itertools.chain.from_iterable(getattr(c, '__slots__', []) for c in cls.__mro__)
        for key in slots:
            setattr(obj, key, deepcopy(getattr(self, key), memo))

        obj.reset()
        return obj

    @abstractmethod
    def increase(self) -> bool:
        """
        Increase the timer value.

        :return: True if timer was increased, False if not.
        """
        raise NotImplementedError

    def reset(self) -> None:
        """Reset the timer to its initial settings."""
        self._value = self._initial
        self._counter = 0

    def wait(self) -> None:
        """Sleep for the current time set for this timer."""
        sleep(self._value)

As an example, the following implements the StepCountTimer.

class CountTimer(Timer, metaclass=ABCMeta):
    """
    Abstract implementation of a :py:class:`Timer` which will increment a maximum number of times.

    :param initial: The starting value to use.
    :param count: The amount of times to increase the value.
    """

    __slots__ = ("_count",)

    @property
    def count(self):
        return self._count

    @property
    def can_increase(self) -> bool:
        return self.count is None or isinstance(self.count, int) and self.counter < self.count

    def __init__(self, initial: Number = 1, count: int = None):
        super().__init__(initial=initial)
        self._count = count

class StepCountTimer(CountTimer):
    """
    Increases timer value by a given ``step`` amount a distinct number of times.

    :param initial: The starting value to use.
    :param count: The amount of times to increase the value.
    :param step: The amount to increase the value by for each value increase.
    """

    __slots__ = ("_step",)

    @property
    def final(self):
        if self.count is None:
            return
        return self.initial + (self.step * self.count)

    @property
    def total(self):
        if self.count is None:
            return
        return sum(self.initial + self.step * i for i in range(self.count + 1))

    @property
    def total_remaining(self):
        if self.count is None:
            return
        return sum(float(self) + self.step * i for i in range(self.count_remaining + 1)) - float(self)

    @property
    def step(self) -> Number:
        """The amount to increase the timer value by in seconds."""
        return self._step

    def __init__(self, initial: Number = 0, count: int = None, step: Number = 1):
        super().__init__(initial=initial, count=count)
        self._step = step

    def increase(self) -> bool:
        if not self.can_increase:
            return False

        self._value += self.step
        self._counter += 1
        return True