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

#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2022
# 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 the Builder classes for the telegram.ext module."""
from asyncio import Queue
from pathlib import Path
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Coroutine,
    Dict,
    Generic,
    Optional,
    Type,
    TypeVar,
    Union,
)

from telegram._bot import Bot
from telegram._utils.defaultvalue import DEFAULT_FALSE, DEFAULT_NONE, DefaultValue
from telegram._utils.types import DVInput, FilePathInput, ODVInput
from telegram.ext._application import Application
from telegram.ext._contexttypes import ContextTypes
from telegram.ext._extbot import ExtBot
from telegram.ext._jobqueue import JobQueue
from telegram.ext._updater import Updater
from telegram.ext._utils.types import BD, BT, CCT, CD, JQ, UD
from telegram.request import BaseRequest
from telegram.request._httpxrequest import HTTPXRequest

if TYPE_CHECKING:
    from telegram.ext import BasePersistence, CallbackContext, Defaults

# Type hinting is a bit complicated here because we try to get to a sane level of
# leveraging generics and therefore need a number of type variables.
# 'In' stands for input - used in parameters of methods below
# pylint: disable=invalid-name
InBT = TypeVar("InBT", bound=Bot)
InJQ = TypeVar("InJQ", bound=Union[None, JobQueue])
InCCT = TypeVar("InCCT", bound="CallbackContext")
InUD = TypeVar("InUD")
InCD = TypeVar("InCD")
InBD = TypeVar("InBD")
BuilderType = TypeVar("BuilderType", bound="ApplicationBuilder")


_BOT_CHECKS = [
    ("request", "request instance"),
    ("get_updates_request", "get_updates_request instance"),
    ("connection_pool_size", "connection_pool_size"),
    ("proxy_url", "proxy_url"),
    ("pool_timeout", "pool_timeout"),
    ("connect_timeout", "connect_timeout"),
    ("read_timeout", "read_timeout"),
    ("write_timeout", "write_timeout"),
    ("get_updates_connection_pool_size", "get_updates_connection_pool_size"),
    ("get_updates_proxy_url", "get_updates_proxy_url"),
    ("get_updates_pool_timeout", "get_updates_pool_timeout"),
    ("get_updates_connect_timeout", "get_updates_connect_timeout"),
    ("get_updates_read_timeout", "get_updates_read_timeout"),
    ("get_updates_write_timeout", "get_updates_write_timeout"),
    ("base_file_url", "base_file_url"),
    ("base_url", "base_url"),
    ("token", "token"),
    ("defaults", "defaults"),
    ("arbitrary_callback_data", "arbitrary_callback_data"),
    ("private_key", "private_key"),
]

_TWO_ARGS_REQ = "The parameter `{}` may only be set, if no {} was set."


class ApplicationBuilder(Generic[BT, CCT, UD, CD, BD, JQ]):
    """This class serves as initializer for :class:`telegram.ext.Application` via the so called
    `builder pattern`_. To build a :class:`telegram.ext.Application`, one first initializes an
    instance of this class. Arguments for the :class:`telegram.ext.Application` to build are then
    added by subsequently calling the methods of the builder. Finally, the
    :class:`telegram.ext.Application` is built by calling :meth:`build`. In the simplest case this
    can look like the following example.

    Example:
        .. code:: python

            application = ApplicationBuilder().token("TOKEN").build()

    Please see the description of the individual methods for information on which arguments can be
    set and what the defaults are when not called. When no default is mentioned, the argument will
    not be used by default.

    Note:
        * Some arguments are mutually exclusive. E.g. after calling :meth:`token`, you can't set
          a custom bot with :meth:`bot` and vice versa.
        * Unless a custom :class:`telegram.Bot` instance is set via :meth:`bot`, :meth:`build` will
          use :class:`telegram.ext.ExtBot` for the bot.

    .. _`builder pattern`: https://en.wikipedia.org/wiki/Builder_pattern
    """

    __slots__ = (
        "_application_class",
        "_application_kwargs",
        "_arbitrary_callback_data",
        "_base_file_url",
        "_base_url",
        "_bot",
        "_concurrent_updates",
        "_connect_timeout",
        "_connection_pool_size",
        "_context_types",
        "_defaults",
        "_get_updates_connect_timeout",
        "_get_updates_connection_pool_size",
        "_get_updates_pool_timeout",
        "_get_updates_proxy_url",
        "_get_updates_read_timeout",
        "_get_updates_request",
        "_get_updates_write_timeout",
        "_job_queue",
        "_persistence",
        "_pool_timeout",
        "_post_init",
        "_private_key",
        "_private_key_password",
        "_proxy_url",
        "_read_timeout",
        "_request",
        "_token",
        "_update_queue",
        "_updater",
        "_write_timeout",
    )

    def __init__(self: "InitApplicationBuilder"):
        self._token: DVInput[str] = DefaultValue("")
        self._base_url: DVInput[str] = DefaultValue("https://api.telegram.org/bot")
        self._base_file_url: DVInput[str] = DefaultValue("https://api.telegram.org/file/bot")
        self._connection_pool_size: DVInput[int] = DEFAULT_NONE
        self._proxy_url: DVInput[str] = DEFAULT_NONE
        self._connect_timeout: ODVInput[float] = DEFAULT_NONE
        self._read_timeout: ODVInput[float] = DEFAULT_NONE
        self._write_timeout: ODVInput[float] = DEFAULT_NONE
        self._pool_timeout: ODVInput[float] = DEFAULT_NONE
        self._request: DVInput["BaseRequest"] = DEFAULT_NONE
        self._get_updates_connection_pool_size: DVInput[int] = DEFAULT_NONE
        self._get_updates_proxy_url: DVInput[str] = DEFAULT_NONE
        self._get_updates_connect_timeout: ODVInput[float] = DEFAULT_NONE
        self._get_updates_read_timeout: ODVInput[float] = DEFAULT_NONE
        self._get_updates_write_timeout: ODVInput[float] = DEFAULT_NONE
        self._get_updates_pool_timeout: ODVInput[float] = DEFAULT_NONE
        self._get_updates_request: DVInput["BaseRequest"] = DEFAULT_NONE
        self._private_key: ODVInput[bytes] = DEFAULT_NONE
        self._private_key_password: ODVInput[bytes] = DEFAULT_NONE
        self._defaults: ODVInput["Defaults"] = DEFAULT_NONE
        self._arbitrary_callback_data: DVInput[Union[bool, int]] = DEFAULT_FALSE
        self._bot: DVInput[Bot] = DEFAULT_NONE
        self._update_queue: DVInput[Queue] = DefaultValue(Queue())
        self._job_queue: ODVInput["JobQueue"] = DefaultValue(JobQueue())
        self._persistence: ODVInput["BasePersistence"] = DEFAULT_NONE
        self._context_types: DVInput[ContextTypes] = DefaultValue(ContextTypes())
        self._application_class: DVInput[Type[Application]] = DefaultValue(Application)
        self._application_kwargs: Dict[str, object] = {}
        self._concurrent_updates: DVInput[Union[int, bool]] = DEFAULT_FALSE
        self._updater: ODVInput[Updater] = DEFAULT_NONE
        self._post_init: Optional[Callable[[Application], Coroutine[Any, Any, None]]] = None

    def _build_request(self, get_updates: bool) -> BaseRequest:
        prefix = "_get_updates_" if get_updates else "_"
        if not isinstance(getattr(self, f"{prefix}request"), DefaultValue):
            return getattr(self, f"{prefix}request")

        proxy_url = DefaultValue.get_value(getattr(self, f"{prefix}proxy_url"))
        if get_updates:
            connection_pool_size = (
                DefaultValue.get_value(getattr(self, f"{prefix}connection_pool_size")) or 1
            )
        else:
            connection_pool_size = (
                DefaultValue.get_value(getattr(self, f"{prefix}connection_pool_size")) or 128
            )

        timeouts = dict(
            connect_timeout=getattr(self, f"{prefix}connect_timeout"),
            read_timeout=getattr(self, f"{prefix}read_timeout"),
            write_timeout=getattr(self, f"{prefix}write_timeout"),
            pool_timeout=getattr(self, f"{prefix}pool_timeout"),
        )
        # Get timeouts that were actually set-
        effective_timeouts = {
            key: value for key, value in timeouts.items() if not isinstance(value, DefaultValue)
        }

        return HTTPXRequest(
            connection_pool_size=connection_pool_size,
            proxy_url=proxy_url,
            **effective_timeouts,
        )

    def _build_ext_bot(self) -> ExtBot:
        if isinstance(self._token, DefaultValue):
            raise RuntimeError("No bot token was set.")

        return ExtBot(
            token=self._token,
            base_url=DefaultValue.get_value(self._base_url),
            base_file_url=DefaultValue.get_value(self._base_file_url),
            private_key=DefaultValue.get_value(self._private_key),
            private_key_password=DefaultValue.get_value(self._private_key_password),
            defaults=DefaultValue.get_value(self._defaults),
            arbitrary_callback_data=DefaultValue.get_value(self._arbitrary_callback_data),
            request=self._build_request(get_updates=False),
            get_updates_request=self._build_request(get_updates=True),
        )

    def build(
        self: "ApplicationBuilder[BT, CCT, UD, CD, BD, JQ]",
    ) -> Application[BT, CCT, UD, CD, BD, JQ]:
        """Builds a :class:`telegram.ext.Application` with the provided arguments.

        Calls :meth:`telegram.ext.JobQueue.set_application` and
        :meth:`telegram.ext.BasePersistence.set_bot` if appropriate.

        Returns:
            :class:`telegram.ext.Application`
        """
        job_queue = DefaultValue.get_value(self._job_queue)
        persistence = DefaultValue.get_value(self._persistence)
        # If user didn't set updater
        if isinstance(self._updater, DefaultValue) or self._updater is None:
            if isinstance(self._bot, DefaultValue):  # and didn't set a bot
                bot: Bot = self._build_ext_bot()  # build a bot
            else:
                bot = self._bot
            # now also build an updater/update_queue for them
            update_queue = DefaultValue.get_value(self._update_queue)

            if self._updater is None:
                updater = None
            else:
                updater = Updater(bot=bot, update_queue=update_queue)
        else:  # if they set an updater, get all necessary attributes for Application from Updater:
            updater = self._updater
            bot = self._updater.bot
            update_queue = self._updater.update_queue

        application: Application[
            BT, CCT, UD, CD, BD, JQ
        ] = DefaultValue.get_value(  # pylint: disable=not-callable
            self._application_class
        )(
            bot=bot,
            update_queue=update_queue,
            updater=updater,
            concurrent_updates=DefaultValue.get_value(self._concurrent_updates),
            job_queue=job_queue,
            persistence=persistence,
            context_types=DefaultValue.get_value(self._context_types),
            post_init=self._post_init,
            **self._application_kwargs,  # For custom Application subclasses
        )

        if job_queue is not None:
            job_queue.set_application(application)

        if persistence is not None:
            # This raises an exception if persistence.store_data.callback_data is True
            # but self.bot is not an instance of ExtBot - so no need to check that later on
            persistence.set_bot(bot)

        return application

    def application_class(
        self: BuilderType, application_class: Type[Application], kwargs: Dict[str, object] = None
    ) -> BuilderType:
        """Sets a custom subclass instead of :class:`telegram.ext.Application`. The
        subclass's ``__init__`` should look like this

        .. code:: python

            def __init__(self, custom_arg_1, custom_arg_2, ..., **kwargs):
                super().__init__(**kwargs)
                self.custom_arg_1 = custom_arg_1
                self.custom_arg_2 = custom_arg_2

        Args:
            application_class (:obj:`type`): A subclass of :class:`telegram.ext.Application`
            kwargs (Dict[:obj:`str`, :obj:`object`], optional): Keyword arguments for the
                initialization. Defaults to an empty dict.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._application_class = application_class
        self._application_kwargs = kwargs or {}
        return self

    def token(self: BuilderType, token: str) -> BuilderType:
        """Sets the token for :attr:`telegram.ext.Application.bot`.

        Args:
            token (:obj:`str`): The token.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("token", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("token", "updater"))
        self._token = token
        return self

    def base_url(self: BuilderType, base_url: str) -> BuilderType:
        """Sets the base URL for :attr:`telegram.ext.Application.bot`. If not called,
        will default to ``'https://api.telegram.org/bot'``.

        .. seealso:: :paramref:`telegram.Bot.base_url`, `Local Bot API Server <https://github.com/\
            python-telegram-bot/python-telegram-bot/wiki/Local-Bot-API-Server>`_,
            :meth:`base_file_url`

        Args:
            base_url (:obj:`str`): The URL.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("base_url", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("base_url", "updater"))
        self._base_url = base_url
        return self

    def base_file_url(self: BuilderType, base_file_url: str) -> BuilderType:
        """Sets the base file URL for :attr:`telegram.ext.Application.bot`. If not
        called, will default to ``'https://api.telegram.org/file/bot'``.

        .. seealso:: :paramref:`telegram.Bot.base_file_url`, `Local Bot API Server <https://\
            github.com/python-telegram-bot/python-telegram-bot/wiki/Local-Bot-API-Server>`_,
            :meth:`base_url`

        Args:
            base_file_url (:obj:`str`): The URL.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("base_file_url", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("base_file_url", "updater"))
        self._base_file_url = base_file_url
        return self

    def _request_check(self, get_updates: bool) -> None:
        prefix = "get_updates_" if get_updates else ""
        name = prefix + "request"

        # Code below tests if it's okay to set a Request object. Only okay if no other request args
        # or instances containing a Request were set previously
        for attr in ("connect_timeout", "read_timeout", "write_timeout", "pool_timeout"):
            if not isinstance(getattr(self, f"_{prefix}{attr}"), DefaultValue):
                raise RuntimeError(_TWO_ARGS_REQ.format(name, attr))
        if not isinstance(getattr(self, f"_{prefix}connection_pool_size"), DefaultValue):
            raise RuntimeError(_TWO_ARGS_REQ.format(name, "connection_pool_size"))
        if not isinstance(getattr(self, f"_{prefix}proxy_url"), DefaultValue):
            raise RuntimeError(_TWO_ARGS_REQ.format(name, "proxy_url"))
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format(name, "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format(name, "updater instance"))

    def _request_param_check(self, name: str, get_updates: bool) -> None:
        if get_updates and self._get_updates_request is not DEFAULT_NONE:
            raise RuntimeError(  # disallow request args for get_updates if Request for that is set
                _TWO_ARGS_REQ.format(f"get_updates_{name}", "get_updates_request instance")
            )
        if self._request is not DEFAULT_NONE:  # disallow request args if request is set
            raise RuntimeError(_TWO_ARGS_REQ.format(name, "request instance"))

        if self._bot is not DEFAULT_NONE:  # disallow request args if bot is set (has Request)
            raise RuntimeError(
                _TWO_ARGS_REQ.format(
                    f"get_updates_{name}" if get_updates else name, "bot instance"
                )
            )

        if self._updater not in (DEFAULT_NONE, None):  # disallow request args for updater(has bot)
            raise RuntimeError(
                _TWO_ARGS_REQ.format(f"get_updates_{name}" if get_updates else name, "updater")
            )

    def request(self: BuilderType, request: BaseRequest) -> BuilderType:
        """Sets a :class:`telegram.request.BaseRequest` instance for the
        :paramref:`telegram.Bot.request` parameter of :attr:`telegram.ext.Application.bot`.

        .. seealso:: :meth:`get_updates_request`

        Args:
            request (:class:`telegram.request.BaseRequest`): The request instance.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_check(get_updates=False)
        self._request = request
        return self

    def connection_pool_size(self: BuilderType, connection_pool_size: int) -> BuilderType:
        """Sets the size of the connection pool for the
        :paramref:`~telegram.request.HTTPXRequest.connection_pool_size` parameter of
        :attr:`telegram.Bot.request`. Defaults to ``128``.

        Args:
            connection_pool_size (:obj:`int`): The size of the connection pool.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="connection_pool_size", get_updates=False)
        self._connection_pool_size = connection_pool_size
        return self

    def proxy_url(self: BuilderType, proxy_url: str) -> BuilderType:
        """Sets the proxy for the :paramref:`~telegram.request.HTTPXRequest.proxy_url`
        parameter of :attr:`telegram.Bot.request`. Defaults to :obj:`None`.

        Args:
            proxy_url (:obj:`str`): The URL to the proxy server. See
                :paramref:`telegram.request.HTTPXRequest.proxy_url` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="proxy_url", get_updates=False)
        self._proxy_url = proxy_url
        return self

    def connect_timeout(self: BuilderType, connect_timeout: Optional[float]) -> BuilderType:
        """Sets the connection attempt timeout for the
        :paramref:`~telegram.request.HTTPXRequest.connect_timeout` parameter of
        :attr:`telegram.Bot.request`. Defaults to ``5.0``.

        Args:
            connect_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.connect_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="connect_timeout", get_updates=False)
        self._connect_timeout = connect_timeout
        return self

    def read_timeout(self: BuilderType, read_timeout: Optional[float]) -> BuilderType:
        """Sets the waiting timeout for the
        :paramref:`~telegram.request.HTTPXRequest.read_timeout` parameter of
        :attr:`telegram.Bot.request`. Defaults to ``5.0``.

        Args:
            read_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.read_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="read_timeout", get_updates=False)
        self._read_timeout = read_timeout
        return self

    def write_timeout(self: BuilderType, write_timeout: Optional[float]) -> BuilderType:
        """Sets the write operation timeout for the
        :paramref:`~telegram.request.HTTPXRequest.write_timeout` parameter of
        :attr:`telegram.Bot.request`. Defaults to ``5.0``.

        Args:
            write_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.write_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="write_timeout", get_updates=False)
        self._write_timeout = write_timeout
        return self

    def pool_timeout(self: BuilderType, pool_timeout: Optional[float]) -> BuilderType:
        """Sets the connection pool's connection freeing timeout for the
        :paramref:`~telegram.request.HTTPXRequest.pool_timeout` parameter of
        :attr:`telegram.Bot.request`. Defaults to :obj:`None`.

        Args:
            pool_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.pool_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="pool_timeout", get_updates=False)
        self._pool_timeout = pool_timeout
        return self

    def get_updates_request(self: BuilderType, get_updates_request: BaseRequest) -> BuilderType:
        """Sets a :class:`telegram.request.BaseRequest` instance for the
        :paramref:`~telegram.Bot.get_updates_request` parameter of
        :attr:`telegram.ext.Application.bot`.

        .. seealso:: :meth:`request`

        Args:
            get_updates_request (:class:`telegram.request.BaseRequest`): The request instance.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_check(get_updates=True)
        self._get_updates_request = get_updates_request
        return self

    def get_updates_connection_pool_size(
        self: BuilderType, get_updates_connection_pool_size: int
    ) -> BuilderType:
        """Sets the size of the connection pool for the
        :paramref:`telegram.request.HTTPXRequest.connection_pool_size` parameter which is used
        for the :meth:`telegram.Bot.get_updates` request. Defaults to ``1``.

        Args:
            get_updates_connection_pool_size (:obj:`int`): The size of the connection pool.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="connection_pool_size", get_updates=True)
        self._get_updates_connection_pool_size = get_updates_connection_pool_size
        return self

    def get_updates_proxy_url(self: BuilderType, get_updates_proxy_url: str) -> BuilderType:
        """Sets the proxy for the :paramref:`telegram.request.HTTPXRequest.proxy_url`
        parameter which is used for :meth:`telegram.Bot.get_updates`. Defaults to :obj:`None`.

        Args:
            get_updates_proxy_url (:obj:`str`): The URL to the proxy server. See
                :paramref:`telegram.request.HTTPXRequest.proxy_url` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="proxy_url", get_updates=True)
        self._get_updates_proxy_url = get_updates_proxy_url
        return self

    def get_updates_connect_timeout(
        self: BuilderType, get_updates_connect_timeout: Optional[float]
    ) -> BuilderType:
        """Sets the connection attempt timeout for the
        :paramref:`telegram.request.HTTPXRequest.connect_timeout` parameter which is used for
        the :meth:`telegram.Bot.get_updates` request. Defaults to ``5.0``.

        Args:
            get_updates_connect_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.connect_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="connect_timeout", get_updates=True)
        self._get_updates_connect_timeout = get_updates_connect_timeout
        return self

    def get_updates_read_timeout(
        self: BuilderType, get_updates_read_timeout: Optional[float]
    ) -> BuilderType:
        """Sets the waiting timeout for the
        :paramref:`telegram.request.HTTPXRequest.read_timeout` parameter which is used for the
        :meth:`telegram.Bot.get_updates` request. Defaults to ``5.0``.

        Args:
            get_updates_read_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.read_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="read_timeout", get_updates=True)
        self._get_updates_read_timeout = get_updates_read_timeout
        return self

    def get_updates_write_timeout(
        self: BuilderType, get_updates_write_timeout: Optional[float]
    ) -> BuilderType:
        """Sets the write operation timeout for the
        :paramref:`telegram.request.HTTPXRequest.write_timeout` parameter which is used for
        the :meth:`telegram.Bot.get_updates` request. Defaults to ``5.0``.

        Args:
            get_updates_write_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.write_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="write_timeout", get_updates=True)
        self._get_updates_write_timeout = get_updates_write_timeout
        return self

    def get_updates_pool_timeout(
        self: BuilderType, get_updates_pool_timeout: Optional[float]
    ) -> BuilderType:
        """Sets the connection pool's connection freeing timeout for the
        :paramref:`~telegram.request.HTTPXRequest.pool_timeout` parameter which is used for the
        :meth:`telegram.Bot.get_updates` request. Defaults to :obj:`None`.

        Args:
            get_updates_pool_timeout (:obj:`float`): See
                :paramref:`telegram.request.HTTPXRequest.pool_timeout` for more information.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._request_param_check(name="pool_timeout", get_updates=True)
        self._get_updates_pool_timeout = get_updates_pool_timeout
        return self

    def private_key(
        self: BuilderType,
        private_key: Union[bytes, FilePathInput],
        password: Union[bytes, FilePathInput] = None,
    ) -> BuilderType:
        """Sets the private key and corresponding password for decryption of telegram passport data
        for :attr:`telegram.ext.Application.bot`.

        .. seealso:: :any:`examples.passportbot`, `Telegram Passports
            <https://github.com/python-telegram-bot/python-telegram-bot/wiki/Telegram-Passport>`_

        Args:
            private_key (:obj:`bytes` | :obj:`str` | :obj:`pathlib.Path`): The private key or the
                file path of a file that contains the key. In the latter case, the file's content
                will be read automatically.
            password (:obj:`bytes` | :obj:`str` | :obj:`pathlib.Path`, optional): The corresponding
                password or the file path of a file that contains the password. In the latter case,
                the file's content will be read automatically.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("private_key", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("private_key", "updater"))

        self._private_key = (
            private_key if isinstance(private_key, bytes) else Path(private_key).read_bytes()
        )
        if password is None or isinstance(password, bytes):
            self._private_key_password = password
        else:
            self._private_key_password = Path(password).read_bytes()

        return self

    def defaults(self: BuilderType, defaults: "Defaults") -> BuilderType:
        """Sets the :class:`telegram.ext.Defaults` instance for
        :attr:`telegram.ext.Application.bot`.

        .. seealso:: `Adding Defaults <https://github.com/python-telegram-bot/python-telegram-bot\
            /wiki/Adding-defaults-to-your-bot>`_

        Args:
            defaults (:class:`telegram.ext.Defaults`): The defaults instance.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("defaults", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("defaults", "updater"))
        self._defaults = defaults
        return self

    def arbitrary_callback_data(
        self: BuilderType, arbitrary_callback_data: Union[bool, int]
    ) -> BuilderType:
        """Specifies whether :attr:`telegram.ext.Application.bot` should allow arbitrary objects as
        callback data for :class:`telegram.InlineKeyboardButton` and how many keyboards should be
        cached in memory. If not called, only strings can be used as callback data and no data will
        be stored in memory.

        .. seealso:: `Arbitrary callback_data <https://github.com/python-telegram-bot\
            /python-telegram-bot/wiki/Arbitrary-callback_data>`_,
            :any:`examples.arbitrarycallbackdatabot`

        Args:
            arbitrary_callback_data (:obj:`bool` | :obj:`int`): If :obj:`True` is passed, the
                default cache size of ``1024`` will be used. Pass an integer to specify a different
                cache size.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._bot is not DEFAULT_NONE:
            raise RuntimeError(_TWO_ARGS_REQ.format("arbitrary_callback_data", "bot instance"))
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("arbitrary_callback_data", "updater"))
        self._arbitrary_callback_data = arbitrary_callback_data
        return self

    def bot(
        self: "ApplicationBuilder[BT, CCT, UD, CD, BD, JQ]",
        bot: InBT,
    ) -> "ApplicationBuilder[InBT, CCT, UD, CD, BD, JQ]":
        """Sets a :class:`telegram.Bot` instance for
        :attr:`telegram.ext.Application.bot`. Instances of subclasses like
        :class:`telegram.ext.ExtBot` are also valid.

        Args:
            bot (:class:`telegram.Bot`): The bot.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("bot", "updater"))
        for attr, error in _BOT_CHECKS:
            if not isinstance(getattr(self, f"_{attr}"), DefaultValue):
                raise RuntimeError(_TWO_ARGS_REQ.format("bot", error))
        self._bot = bot
        return self  # type: ignore[return-value]

    def update_queue(self: BuilderType, update_queue: Queue) -> BuilderType:
        """Sets a :class:`asyncio.Queue` instance for
        :attr:`telegram.ext.Application.update_queue`, i.e. the queue that the application will
        fetch updates from. Will also be used for the :attr:`telegram.ext.Application.updater`.
        If not called, a queue will be instantiated.

        .. seealso:: :attr:`telegram.ext.Updater.update_queue`

        Args:
            update_queue (:class:`asyncio.Queue`): The queue.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if self._updater not in (DEFAULT_NONE, None):
            raise RuntimeError(_TWO_ARGS_REQ.format("update_queue", "updater instance"))
        self._update_queue = update_queue
        return self

    def concurrent_updates(self: BuilderType, concurrent_updates: Union[bool, int]) -> BuilderType:
        """Specifies if and how many updates may be processed concurrently instead of one by one.

        Warning:
            Processing updates concurrently is not recommended when stateful handlers like
            :class:`telegram.ext.ConversationHandler` are used. Only use this if you are sure
            that your bot does not (explicitly or implicitly) rely on updates being processed
            sequentially.

        .. seealso:: :attr:`telegram.ext.Application.concurrent_updates`

        Args:
            concurrent_updates (:obj:`bool` | :obj:`int`): Passing :obj:`True` will allow for
                ``4096`` updates to be processed concurrently. Pass an integer to specify a
                different number of updates that may be processed concurrently.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._concurrent_updates = concurrent_updates
        return self

    def job_queue(
        self: "ApplicationBuilder[BT, CCT, UD, CD, BD, JQ]",
        job_queue: InJQ,
    ) -> "ApplicationBuilder[BT, CCT, UD, CD, BD, InJQ]":
        """Sets a :class:`telegram.ext.JobQueue` instance for
        :attr:`telegram.ext.Application.job_queue`. If not called, a job queue will be
        instantiated.

        .. seealso:: `JobQueue <https://github.com/python-telegram-bot/python-telegram-bot/wiki\
            /Extensions-%E2%80%93-JobQueue>`_, :any:`examples.timerbot`

        Note:
            * :meth:`telegram.ext.JobQueue.set_application` will be called automatically by
              :meth:`build`.
            * The job queue will be automatically started and stopped by
              :meth:`telegram.ext.Application.start` and :meth:`telegram.ext.Application.stop`,
              respectively.
            * When passing :obj:`None`,
              :attr:`telegram.ext.ConversationHandler.conversation_timeout` can not be used, as
              this uses :attr:`telegram.ext.Application.job_queue` internally.

        Args:
            job_queue (:class:`telegram.ext.JobQueue`): The job queue. Pass :obj:`None` if you
                don't want to use a job queue.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._job_queue = job_queue
        return self  # type: ignore[return-value]

    def persistence(self: BuilderType, persistence: "BasePersistence") -> BuilderType:
        """Sets a :class:`telegram.ext.BasePersistence` instance for
        :attr:`telegram.ext.Application.persistence`.

        Note:
            When using a persistence, note that all
            data stored in :attr:`context.user_data <telegram.ext.CallbackContext.user_data>`,
            :attr:`context.chat_data <telegram.ext.CallbackContext.chat_data>`,
            :attr:`context.bot_data <telegram.ext.CallbackContext.bot_data>` and
            in :attr:`telegram.ext.ExtBot.callback_data_cache` must be copyable with
            :func:`copy.deepcopy`. This is due to the data being deep copied before handing it over
            to the persistence in order to avoid race conditions.

        .. seealso:: `Making your bot persistent <https://github.com/python-telegram-bot\
            /python-telegram-bot/wiki/Making-your-bot-persistent>`_,
            :any:`examples.persistentconversationbot`

        Warning:
            If a :class:`telegram.ext.ContextTypes` instance is set via :meth:`context_types`,
            the persistence instance must use the same types!

        Args:
            persistence (:class:`telegram.ext.BasePersistence`): The persistence instance.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._persistence = persistence
        return self

    def context_types(
        self: "ApplicationBuilder[BT, CCT, UD, CD, BD, JQ]",
        context_types: "ContextTypes[InCCT, InUD, InCD, InBD]",
    ) -> "ApplicationBuilder[BT, InCCT, InUD, InCD, InBD, JQ]":
        """Sets a :class:`telegram.ext.ContextTypes` instance for
        :attr:`telegram.ext.Application.context_types`.

        .. seealso:: :any:`examples.contexttypesbot`

        Args:
            context_types (:class:`telegram.ext.ContextTypes`): The context types.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._context_types = context_types
        return self  # type: ignore[return-value]

    def updater(self: BuilderType, updater: Optional[Updater]) -> BuilderType:
        """Sets a :class:`telegram.ext.Updater` instance for
        :attr:`telegram.ext.Application.updater`. The :attr:`telegram.ext.Updater.bot` and
        :attr:`telegram.ext.Updater.update_queue` will be used for
        :attr:`telegram.ext.Application.bot` and :attr:`telegram.ext.Application.update_queue`,
        respectively.

        Args:
            updater (:class:`telegram.ext.Updater` | :obj:`None`): The updater instance or
                :obj:`None` if no updater should be used.

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        if updater is None:
            self._updater = updater
            return self

        for attr, error in (
            (self._bot, "bot instance"),
            (self._update_queue, "update_queue"),
        ):
            if not isinstance(attr, DefaultValue):
                raise RuntimeError(_TWO_ARGS_REQ.format("updater", error))

        for attr_name, error in _BOT_CHECKS:
            if not isinstance(getattr(self, f"_{attr_name}"), DefaultValue):
                raise RuntimeError(_TWO_ARGS_REQ.format("updater", error))

        self._updater = updater
        return self

    def post_init(
        self: BuilderType, post_init: Callable[[Application], Coroutine[Any, Any, None]]
    ) -> BuilderType:
        """
        Sets a callback to be executed by :meth:`Application.run_polling` and
        :meth:`Application.run_webhook` *after* executing :meth:`Application.initialize` but
        *before* executing :meth:`Updater.start_polling` or :meth:`Updater.start_webhook`,
        respectively.

        Tip:
            This can be used for custom startup logic that requires to await coroutines, e.g.
            setting up the bots commands via :meth:`~telegram.Bot.set_my_commands`.

        Example:
            .. code::

                async def post_init(application: Application) -> None:
                    await application.bot.set_my_commands([('start', 'Starts the bot')])

                application = Application.builder().token("TOKEN").post_init(post_init).build()

        Args:
            post_init (:term:`coroutine function`): The custom callback. Must be a
                :term:`coroutine function` and must accept exactly one positional argument, which
                is the :class:`~telegram.ext.Application`::

                    async def post_init(application: Application) -> None:

        Returns:
            :class:`ApplicationBuilder`: The same builder with the updated argument.
        """
        self._post_init = post_init
        return self


InitApplicationBuilder = (  # This is defined all the way down here so that its type is inferred
    ApplicationBuilder[  # by Pylance correctly.
        ExtBot,
        ContextTypes.DEFAULT_TYPE,
        Dict,
        Dict,
        Dict,
        JobQueue,
    ]
)