python-telegram-bot/telegram/ext/_baseratelimiter.py

#!/usr/bin/env python
#
#  A library that provides a Python interface to the Telegram Bot API
#  Copyright (C) 2015-2024
#  Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU Lesser Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU Lesser Public License for more details.
#
#  You should have received a copy of the GNU Lesser Public License
#  along with this program.  If not, see [http://www.gnu.org/licenses/].
"""This module contains a class that allows to rate limit requests to the Bot API."""
from abc import ABC, abstractmethod
from typing import Any, Callable, Coroutine, Dict, Generic, List, Optional, Union

from telegram._utils.types import JSONDict
from telegram.ext._utils.types import RLARGS


class BaseRateLimiter(ABC, Generic[RLARGS]):
    """
    Abstract interface class that allows to rate limit the requests that python-telegram-bot
    sends to the Telegram Bot API. An implementation of this class
    must implement all abstract methods and properties.

    This class is a :class:`~typing.Generic` class and accepts one type variable that specifies
    the type of the argument :paramref:`~process_request.rate_limit_args` of
    :meth:`process_request` and the methods of :class:`~telegram.ext.ExtBot`.

    Hint:
        Requests to :meth:`~telegram.Bot.get_updates` are never rate limited.

    .. seealso:: :wiki:`Architecture Overview <Architecture>`,
        :wiki:`Avoiding Flood Limits <Avoiding-flood-limits>`

    .. versionadded:: 20.0
    """

    __slots__ = ()

    @abstractmethod
    async def initialize(self) -> None:
        """Initialize resources used by this class. Must be implemented by a subclass."""

    @abstractmethod
    async def shutdown(self) -> None:
        """Stop & clear resources used by this class. Must be implemented by a subclass."""

    @abstractmethod
    async def process_request(
        self,
        callback: Callable[..., Coroutine[Any, Any, Union[bool, JSONDict, List[JSONDict]]]],
        args: Any,
        kwargs: Dict[str, Any],
        endpoint: str,
        data: Dict[str, Any],
        rate_limit_args: Optional[RLARGS],
    ) -> Union[bool, JSONDict, List[JSONDict]]:
        """
        Process a request. Must be implemented by a subclass.

        This method must call :paramref:`callback` and return the result of the call.
        `When` the callback is called is up to the implementation.

        Important:
            This method must only return once the result of :paramref:`callback` is known!

        If a :exc:`~telegram.error.RetryAfter` error is raised, this method may try to make
        a new request by calling the callback again.

        Warning:
            This method *should not* handle any other exception raised by :paramref:`callback`!

        There are basically two different approaches how a rate limiter can be implemented:

        1. React only if necessary. In this case, the :paramref:`callback` is called without any
           precautions. If a :exc:`~telegram.error.RetryAfter` error is raised, processing requests
           is halted for the :attr:`~telegram.error.RetryAfter.retry_after` and finally the
           :paramref:`callback` is called again. This approach is often amendable for bots that
           don't have a large user base and/or don't send more messages than they get updates.
        2. Throttle all outgoing requests. In this case the implementation makes sure that the
           requests are spread out over a longer time interval in order to stay below the rate
           limits. This approach is often amendable for bots that have a large user base and/or
           send more messages than they get updates.

        An implementation can use the information provided by :paramref:`data`,
        :paramref:`endpoint` and :paramref:`rate_limit_args` to handle each request differently.

        Examples:
            * It is usually desirable to call :meth:`telegram.Bot.answer_inline_query`
              as quickly as possible, while delaying :meth:`telegram.Bot.send_message`
              is acceptable.
            * There are `different <https://core.telegram.org/bots/faq\
              #my-bot-is-hitting-limits-how-do-i-avoid-this>`_ rate limits for group chats and
              private chats.
            * When sending broadcast messages to a large number of users, these requests can
              typically be delayed for a longer time than messages that are direct replies to a
              user input.

        Args:
            callback (Callable[..., :term:`coroutine`]): The coroutine function that must be called
                to make the request.
            args (Tuple[:obj:`object`]): The positional arguments for the :paramref:`callback`
                function.
            kwargs (Dict[:obj:`str`, :obj:`object`]): The keyword arguments for the
                :paramref:`callback` function.
            endpoint (:obj:`str`): The endpoint that the request is made for, e.g.
                ``"sendMessage"``.
            data (Dict[:obj:`str`, :obj:`object`]): The parameters that were passed to the method
                of :class:`~telegram.ext.ExtBot`. Any ``api_kwargs`` are included in this and
                any :paramref:`~telegram.ext.ExtBot.defaults` are already applied.

                Example:

                    When calling::

                        await ext_bot.send_message(
                            chat_id=1,
                            text="Hello world!",
                            api_kwargs={"custom": "arg"}
                        )

                    then :paramref:`data` will be::

                        {"chat_id": 1, "text": "Hello world!", "custom": "arg"}

            rate_limit_args (:obj:`None` | :class:`object`): Custom arguments passed to the methods
                of :class:`~telegram.ext.ExtBot`. Can e.g. be used to specify the priority of
                the request.

        Returns:
            :obj:`bool` | Dict[:obj:`str`, :obj:`object`] | :obj:`None`: The result of the
            callback function.
        """