mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-12-22 22:45:09 +01:00
946 lines
41 KiB
Python
946 lines
41 KiB
Python
#!/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:: `passportbot.py <https://github.com/python-telegram-bot/python-telegram-bot\
|
|
/tree/master/examples#passportbotpy>`_, `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>`_,
|
|
`arbitrarycallbackdatabot.py <https://github.com/python-telegram-bot\
|
|
/python-telegram-bot/tree/master/examples#arbitrarycallbackdatabotpy>`_
|
|
|
|
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>`_, `timerbot.py <https://github.com\
|
|
/python-telegram-bot/python-telegram-bot/tree/master/examples#timerbotpy>`_
|
|
|
|
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>`_,
|
|
`persistentconversationbot.py <https://github.com/python-telegram-bot\
|
|
/python-telegram-bot/tree/master/examples#persistentconversationbotpy>`_
|
|
|
|
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:: `contexttypesbot.py <https://github.com/python-telegram-bot\
|
|
/python-telegram-bot/tree/master/examples#contexttypesbotpy>`_
|
|
|
|
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,
|
|
]
|
|
)
|