Tasks

Task loops, adapted from the discord.py library.

ExponentialBackoff

Implements the exponential backoff algorithm.

Provides a convenient interface to implement an exponential backoff for reconnecting or retrying transmissions in a distributed network.

Source code in gd/tasks.py

@define()
class ExponentialBackoff:
    """Implements the *exponential backoff* algorithm.

    Provides a convenient interface to implement an exponential backoff
    for reconnecting or retrying transmissions in a distributed network.
    """

    multiply: float = field(default=DEFAULT_MULTIPLY)
    base: float = field(default=DEFAULT_BASE)
    limit: int = field(default=DEFAULT_LIMIT)

    _clock: Clock = field(default=clock)

    _exponent: int = field(default=0, init=False)
    _last_called: float = field(init=False)
    _reset_delta: float = field(init=False)

    @_last_called.default
    def default_last_called(self) -> float:
        return self._clock()

    @_reset_delta.default
    def default_reset_delta(self) -> float:
        return self.multiply * pow(self.base, self.limit + 1)

    def delay(self) -> float:
        """Computes the next delay."""
        called = self._clock()

        interval = called - self._last_called
        self._last_called = called

        if interval > self._reset_delta:
            self._exponent = 0

        if self._exponent < self.limit:
            self._exponent += 1

        return uniform_to(self.multiply * pow(self.base, self._exponent))

delay() -> float

Computes the next delay.

Source code in gd/tasks.py

def delay(self) -> float:
    """Computes the next delay."""
    called = self._clock()

    interval = called - self._last_called
    self._last_called = called

    if interval > self._reset_delta:
        self._exponent = 0

    if self._exponent < self.limit:
        self._exponent += 1

    return uniform_to(self.multiply * pow(self.base, self._exponent))

Loop

Bases: Generic[P]

Abstracts away event loop handling and reconnection logic.

The main interface to create this is through [loop][gd.tasks.loop].

Source code in gd/tasks.py

@define()
class Loop(Generic[P]):
    """Abstracts away event loop handling and reconnection logic.

    The main interface to create this is through [`loop`][gd.tasks.loop].
    """

    function: AsyncCallable[P, None] = field()

    delay: float = field(default=DEFAULT_DELAY)

    count: Optional[int] = field(default=None)

    reconnect: bool = field(default=DEFAULT_RECONNECT)

    _task: Optional[Task[None]] = field(default=None, init=False)

    _current_count: int = field(default=0, init=False)

    _error_types: AnyErrorTypes = field(
        default=(OSError, GDError, ClientError, TimeoutError),
        init=False,
    )

    _is_being_cancelled: bool = field(default=False, init=False)
    _has_failed: bool = field(default=False, init=False)
    _stop_next_iteration: bool = field(default=False, init=False)

    _injected: Optional[Any] = field(default=None, init=False)

    _before_loop: Optional[LoopFunction[Any]] = field(default=None, init=False)
    _after_loop: Optional[LoopFunction[Any]] = field(default=None, init=False)

    async def _call_before_loop(self) -> None:
        before_loop = self._before_loop

        if before_loop is None:
            return

        injected = self._injected

        if injected is None:
            await before_loop()  # type: ignore

        else:
            await before_loop(injected)  # type: ignore

    async def _call_after_loop(self) -> None:
        after_loop = self._after_loop

        if after_loop is None:
            return

        injected = self._injected

        if injected is None:
            await after_loop()  # type: ignore

        else:
            await after_loop(injected)  # type: ignore

    async def _loop(self, *args: P.args, **kwargs: P.kwargs) -> None:
        backoff = ExponentialBackoff()

        await self._call_before_loop()

        try:
            while True:
                try:
                    await self.function(*args, **kwargs)

                except self._error_types:
                    if not self.reconnect:
                        raise

                    await sleep(backoff.delay())

                else:
                    if self._stop_next_iteration:
                        return

                    self._current_count += 1

                    if self._current_count == self.count:
                        break

                    await sleep(self.delay)

        except CancelledError:
            self._is_being_cancelled = True

            raise

        except NormalError:
            self._has_failed = True

            raise

        finally:
            await self._call_after_loop()

            self._current_count = 0

            self._is_being_cancelled = False
            self._has_failed = False
            self._stop_next_iteration = False

    def __get__(self, instance: Optional[S], type: Optional[Type[S]] = None) -> Self:
        if instance is None:
            return self

        self._injected = instance

        return self

    @property
    def current_count(self) -> int:
        return self._current_count

    def start(self, *args: Any, **kwargs: Any) -> Task[None]:
        task = self._task

        if task is not None and not task.done():
            raise RuntimeError(TASK_ALREADY_LAUNCHED)

        injected = self._injected

        if injected is not None:
            args = (injected, *args)

        self._task = task = get_event_loop().create_task(self._loop(*args, **kwargs))

        return task

    def stop(self) -> None:
        task = self._task

        if task is not None and not task.done():
            self._stop_next_iteration = True

    @property
    def _can_be_cancelled(self) -> bool:
        task = self._task

        return not self._is_being_cancelled and task is not None and not task.done()

    def cancel(self) -> None:
        if self._can_be_cancelled:
            self._task.cancel()  # type: ignore

    def restart(self, *args: Any, **kwargs: Any) -> None:
        def restart_when_over(
            task: Task[None],
            *,
            args: DynamicTuple[Any] = args,
            kwargs: StringDict[Any] = kwargs,
        ) -> None:
            self._task.remove_done_callback(restart_when_over)  # type: ignore
            self.start(*args, **kwargs)

        if self._can_be_cancelled:
            self._task.add_done_callback(restart_when_over)  # type: ignore
            self._task.cancel()  # type: ignore

    def add_error_type(self, error_type: AnyErrorType) -> None:
        self._error_types = (*self._error_types, error_type)

    def clear_error_types(self) -> None:
        self._error_types = ()

    def remove_error_type(self, error_type: AnyErrorType) -> bool:
        error_types = self._error_types

        length = len(error_types)

        self._error_types = error_types = tuple(
            present_error_type
            for present_error_type in error_types
            if present_error_type is not error_type
        )

        return len(error_types) < length

    def is_being_cancelled(self) -> bool:
        return self._is_being_cancelled

    def has_failed(self) -> bool:
        return self._has_failed

    def before_loop(self, loop_function: F) -> F:
        self._before_loop = loop_function

        return loop_function

    def after_loop(self, loop_function: F) -> F:
        self._after_loop = loop_function

        return loop_function