Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Extend Customization Support forBot.base_(file_)url#4632

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
Bibo-Joshi merged 8 commits intomasterfromfeature/test-env
Jan 23, 2025
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 80 additions & 9 deletionstelegram/_bot.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -96,7 +96,14 @@
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.strings import to_camel_case
from telegram._utils.types import CorrectOptionID, FileInput, JSONDict, ODVInput, ReplyMarkup
from telegram._utils.types import (
BaseUrl,
CorrectOptionID,
FileInput,
JSONDict,
ODVInput,
ReplyMarkup,
)
from telegram._utils.warnings import warn
from telegram._webhookinfo import WebhookInfo
from telegram.constants import InlineQueryLimit, ReactionEmoji
Expand DownExpand Up@@ -126,6 +133,35 @@
BT = TypeVar("BT", bound="Bot")


# Even though we document only {token} as supported insertion, we are a bit more flexible
# internally and support additional variants. At the very least, we don't want the insertion
# to be case sensitive.
_SUPPORTED_INSERTIONS = {"token", "TOKEN", "bot_token", "BOT_TOKEN", "bot-token", "BOT-TOKEN"}
_INSERTION_STRINGS = {f"{{{insertion}}}" for insertion in _SUPPORTED_INSERTIONS}


class _TokenDict(dict):
__slots__ = ("token",)

# small helper to make .format_map work without knowing which exact insertion name is used
def __init__(self, token: str):
self.token = token
super().__init__()

def __missing__(self, key: str) -> str:
if key in _SUPPORTED_INSERTIONS:
return self.token
raise KeyError(f"Base URL string contains unsupported insertion: {key}")


def _parse_base_url(value: BaseUrl, token: str) -> str:
if callable(value):
return value(token)
if any(insertion in value for insertion in _INSERTION_STRINGS):
return value.format_map(_TokenDict(token))
return value + token


class Bot(TelegramObject, contextlib.AbstractAsyncContextManager["Bot"]):
"""This object represents a Telegram Bot.

Expand DownExpand Up@@ -193,8 +229,40 @@ class Bot(TelegramObject, contextlib.AbstractAsyncContextManager["Bot"]):

Args:
token (:obj:`str`): Bot's unique authentication token.
base_url (:obj:`str`, optional): Telegram Bot API service URL.
base_url (:obj:`str` | Callable[[:obj:`str`], :obj:`str`], optional): Telegram Bot API
service URL. If the string contains ``{token}``, it will be replaced with the bot's
token. If a callable is passed, it will be called with the bot's token as the only
argument and must return the base URL. Otherwise, the token will be appended to the
string. Defaults to ``"https://api.telegram.org/bot"``.

Tip:
Customizing the base URL can be used to run a bot against
:wiki:`Local Bot API Server <Local-Bot-API-Server>` or using Telegrams
`test environment \
<https://core.telegram.org/bots/features#dedicated-test-environment>`_.

Example:
``"https://api.telegram.org/bot{token}/test"``

.. versionchanged:: NEXT.VERSION
Supports callable input and string formatting.
base_file_url (:obj:`str`, optional): Telegram Bot API file URL.
If the string contains ``{token}``, it will be replaced with the bot's
token. If a callable is passed, it will be called with the bot's token as the only
argument and must return the base URL. Otherwise, the token will be appended to the
string. Defaults to ``"https://api.telegram.org/bot"``.

Tip:
Customizing the base URL can be used to run a bot against
:wiki:`Local Bot API Server <Local-Bot-API-Server>` or using Telegrams
`test environment \
<https://core.telegram.org/bots/features#dedicated-test-environment>`_.

Example:
``"https://api.telegram.org/file/bot{token}/test"``

.. versionchanged:: NEXT.VERSION
Supports callable input and string formatting.
request (:class:`telegram.request.BaseRequest`, optional): Pre initialized
:class:`telegram.request.BaseRequest` instances. Will be used for all bot methods
*except* for :meth:`get_updates`. If not passed, an instance of
Expand DownExpand Up@@ -239,8 +307,8 @@ class Bot(TelegramObject, contextlib.AbstractAsyncContextManager["Bot"]):
def __init__(
self,
token: str,
base_url:str = "https://api.telegram.org/bot",
base_file_url:str = "https://api.telegram.org/file/bot",
base_url:BaseUrl = "https://api.telegram.org/bot",
base_file_url:BaseUrl = "https://api.telegram.org/file/bot",
request: Optional[BaseRequest] = None,
get_updates_request: Optional[BaseRequest] = None,
private_key: Optional[bytes] = None,
Expand All@@ -252,8 +320,11 @@ def __init__(
raise InvalidToken("You must pass the token you received from https://t.me/Botfather!")
self._token: str = token

self._base_url: str = base_url + self._token
self._base_file_url: str = base_file_url + self._token
self._base_url: str = _parse_base_url(base_url, self._token)
self._base_file_url: str = _parse_base_url(base_file_url, self._token)
self._LOGGER.debug("Set Bot API URL: %s", self._base_url)
self._LOGGER.debug("Set Bot API File URL: %s", self._base_file_url)

self._local_mode: bool = local_mode
self._bot_user: Optional[User] = None
self._private_key: Optional[bytes] = None
Expand All@@ -264,7 +335,7 @@ def __init__(
HTTPXRequest() if request is None else request,
)

# this section is about issuing a warning when using HTTP/2 and connect to a selfhosted
# this section is about issuing a warning when using HTTP/2 and connect to a self-hosted
# bot api instance, which currently only supports HTTP/1.1. Checking if a custom base url
# is set is the best way to do that.

Expand All@@ -273,14 +344,14 @@ def __init__(
if (
isinstance(self._request[0], HTTPXRequest)
and self._request[0].http_version == "2"
and not base_url.startswith("https://api.telegram.org/bot")
and notself.base_url.startswith("https://api.telegram.org/bot")
):
warning_string = "get_updates_request"

if (
isinstance(self._request[1], HTTPXRequest)
and self._request[1].http_version == "2"
and not base_url.startswith("https://api.telegram.org/bot")
and notself.base_url.startswith("https://api.telegram.org/bot")
):
if warning_string:
warning_string += " and request"
Expand Down
4 changes: 3 additions & 1 deletiontelegram/_utils/types.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,7 @@
"""
from collections.abc import Collection
from pathlib import Path
from typing import IO, TYPE_CHECKING, Any, Literal, Optional, TypeVar, Union
from typing import IO, TYPE_CHECKING, Any,Callable,Literal, Optional, TypeVar, Union

if TYPE_CHECKING:
from telegram import (
Expand DownExpand Up@@ -91,3 +91,5 @@
tuple[int, int, Union[bytes, bytearray]],
tuple[int, int, None, int],
]

BaseUrl = Union[str, Callable[[str], str]]
30 changes: 23 additions & 7 deletionstelegram/ext/_applicationbuilder.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -26,7 +26,15 @@

from telegram._bot import Bot
from telegram._utils.defaultvalue import DEFAULT_FALSE, DEFAULT_NONE, DefaultValue
from telegram._utils.types import DVInput, DVType, FilePathInput, HTTPVersion, ODVInput, SocketOpt
from telegram._utils.types import (
BaseUrl,
DVInput,
DVType,
FilePathInput,
HTTPVersion,
ODVInput,
SocketOpt,
)
from telegram._utils.warnings import warn
from telegram.ext._application import Application
from telegram.ext._baseupdateprocessor import BaseUpdateProcessor, SimpleUpdateProcessor
Expand DownExpand Up@@ -164,8 +172,8 @@ class ApplicationBuilder(Generic[BT, CCT, UD, CD, BD, JQ]):

def __init__(self: "InitApplicationBuilder"):
self._token: DVType[str] = DefaultValue("")
self._base_url: DVType[str] = DefaultValue("https://api.telegram.org/bot")
self._base_file_url: DVType[str] = DefaultValue("https://api.telegram.org/file/bot")
self._base_url: DVType[BaseUrl] = DefaultValue("https://api.telegram.org/bot")
self._base_file_url: DVType[BaseUrl] = DefaultValue("https://api.telegram.org/file/bot")
self._connection_pool_size: DVInput[int] = DEFAULT_NONE
self._proxy: DVInput[Union[str, httpx.Proxy, httpx.URL]] = DEFAULT_NONE
self._socket_options: DVInput[Collection[SocketOpt]] = DEFAULT_NONE
Expand DownExpand Up@@ -378,15 +386,19 @@ def token(self: BuilderType, token: str) -> BuilderType:
self._token = token
return self

def base_url(self: BuilderType, base_url:str) -> BuilderType:
def base_url(self: BuilderType, base_url:BaseUrl) -> 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`,
:wiki:`Local Bot API Server <Local-Bot-API-Server>`, :meth:`base_file_url`

.. versionchanged:: NEXT.VERSION
Supports callable input and string formatting.

Args:
base_url (:obj:`str`): The URL.
base_url (:obj:`str` | Callable[[:obj:`str`], :obj:`str`]): The URL or
input for the URL as accepted by :paramref:`telegram.Bot.base_url`.

Returns:
:class:`ApplicationBuilder`: The same builder with the updated argument.
Expand All@@ -396,15 +408,19 @@ def base_url(self: BuilderType, base_url: str) -> BuilderType:
self._base_url = base_url
return self

def base_file_url(self: BuilderType, base_file_url:str) -> BuilderType:
def base_file_url(self: BuilderType, base_file_url:BaseUrl) -> 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`,
:wiki:`Local Bot API Server <Local-Bot-API-Server>`, :meth:`base_url`

.. versionchanged:: NEXT.VERSION
Supports callable input and string formatting.

Args:
base_file_url (:obj:`str`): The URL.
base_file_url (:obj:`str` | Callable[[:obj:`str`], :obj:`str`]): The URL or
input for the URL as accepted by :paramref:`telegram.Bot.base_file_url`.

Returns:
:class:`ApplicationBuilder`: The same builder with the updated argument.
Expand Down
21 changes: 14 additions & 7 deletionstelegram/ext/_extbot.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -93,7 +93,14 @@
from telegram._utils.defaultvalue import DEFAULT_NONE, DefaultValue
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import CorrectOptionID, FileInput, JSONDict, ODVInput, ReplyMarkup
from telegram._utils.types import (
BaseUrl,
CorrectOptionID,
FileInput,
JSONDict,
ODVInput,
ReplyMarkup,
)
from telegram.ext._callbackdatacache import CallbackDataCache
from telegram.ext._utils.types import RLARGS
from telegram.request import BaseRequest
Expand DownExpand Up@@ -184,8 +191,8 @@ class ExtBot(Bot, Generic[RLARGS]):
def __init__(
self: "ExtBot[None]",
token: str,
base_url:str = "https://api.telegram.org/bot",
base_file_url:str = "https://api.telegram.org/file/bot",
base_url:BaseUrl = "https://api.telegram.org/bot",
base_file_url:BaseUrl = "https://api.telegram.org/file/bot",
request: Optional[BaseRequest] = None,
get_updates_request: Optional[BaseRequest] = None,
private_key: Optional[bytes] = None,
Expand All@@ -199,8 +206,8 @@ def __init__(
def __init__(
self: "ExtBot[RLARGS]",
token: str,
base_url:str = "https://api.telegram.org/bot",
base_file_url:str = "https://api.telegram.org/file/bot",
base_url:BaseUrl = "https://api.telegram.org/bot",
base_file_url:BaseUrl = "https://api.telegram.org/file/bot",
request: Optional[BaseRequest] = None,
get_updates_request: Optional[BaseRequest] = None,
private_key: Optional[bytes] = None,
Expand All@@ -214,8 +221,8 @@ def __init__(
def __init__(
self,
token: str,
base_url:str = "https://api.telegram.org/bot",
base_file_url:str = "https://api.telegram.org/file/bot",
base_url:BaseUrl = "https://api.telegram.org/bot",
base_file_url:BaseUrl = "https://api.telegram.org/file/bot",
request: Optional[BaseRequest] = None,
get_updates_request: Optional[BaseRequest] = None,
private_key: Optional[bytes] = None,
Expand Down
72 changes: 70 additions & 2 deletionstests/test_bot.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -235,7 +235,9 @@ def _reset(self):

@pytest.mark.parametrize("bot_class", [Bot, ExtBot])
def test_slot_behaviour(self, bot_class, offline_bot):
inst = bot_class(offline_bot.token)
inst = bot_class(
offline_bot.token, request=OfflineRequest(1), get_updates_request=OfflineRequest(1)
)
for attr in inst.__slots__:
assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'"
assert len(mro_slots(inst)) == len(set(mro_slots(inst))), "duplicate slot"
Expand All@@ -244,6 +246,71 @@ async def test_no_token_passed(self):
with pytest.raises(InvalidToken, match="You must pass the token"):
Bot("")

def test_base_url_parsing_basic(self, caplog):
with caplog.at_level(logging.DEBUG):
bot = Bot(
token="!!Test String!!",
base_url="base/",
base_file_url="base/",
request=OfflineRequest(1),
get_updates_request=OfflineRequest(1),
)

assert bot.base_url == "base/!!Test String!!"
assert bot.base_file_url == "base/!!Test String!!"

assert len(caplog.records) >= 2
messages = [record.getMessage() for record in caplog.records]
assert "Set Bot API URL: base/!!Test String!!" in messages
assert "Set Bot API File URL: base/!!Test String!!" in messages

@pytest.mark.parametrize(
"insert_key", ["token", "TOKEN", "bot_token", "BOT_TOKEN", "bot-token", "BOT-TOKEN"]
)
def test_base_url_parsing_string_format(self, insert_key, caplog):
string = f"{{{insert_key}}}"

with caplog.at_level(logging.DEBUG):
bot = Bot(
token="!!Test String!!",
base_url=string,
base_file_url=string,
request=OfflineRequest(1),
get_updates_request=OfflineRequest(1),
)

assert bot.base_url == "!!Test String!!"
assert bot.base_file_url == "!!Test String!!"

assert len(caplog.records) >= 2
messages = [record.getMessage() for record in caplog.records]
assert "Set Bot API URL: !!Test String!!" in messages
assert "Set Bot API File URL: !!Test String!!" in messages

with pytest.raises(KeyError, match="unsupported insertion: unknown"):
Bot("token", base_url="{unknown}{token}")

def test_base_url_parsing_callable(self, caplog):
def build_url(_: str) -> str:
return "!!Test String!!"

with caplog.at_level(logging.DEBUG):
bot = Bot(
token="some-token",
base_url=build_url,
base_file_url=build_url,
request=OfflineRequest(1),
get_updates_request=OfflineRequest(1),
)

assert bot.base_url == "!!Test String!!"
assert bot.base_file_url == "!!Test String!!"

assert len(caplog.records) >= 2
messages = [record.getMessage() for record in caplog.records]
assert "Set Bot API URL: !!Test String!!" in messages
assert "Set Bot API File URL: !!Test String!!" in messages

async def test_repr(self):
offline_bot = Bot(token="some_token", base_file_url="")
assert repr(offline_bot) == "Bot[token=some_token]"
Expand DownExpand Up@@ -409,9 +476,10 @@ def test_bot_deepcopy_error(self, offline_bot):
("cls", "logger_name"), [(Bot, "telegram.Bot"), (ExtBot, "telegram.ext.ExtBot")]
)
async def test_bot_method_logging(self, offline_bot: PytestExtBot, cls, logger_name, caplog):
instance = cls(offline_bot.token)
# Second argument makes sure that we ignore logs from e.g. httpx
with caplog.at_level(logging.DEBUG, logger="telegram"):
awaitcls(offline_bot.token).get_me()
awaitinstance.get_me()
# Only for stabilizing this test-
if len(caplog.records) == 4:
for idx, record in enumerate(caplog.records):
Expand Down
Loading
[8]ページ先頭
©2009-2025 Movatter.jp