OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2025-03-17 04:39:55 +01:00
Code Issues Projects Releases Packages Wiki Activity

Automatic Pagination for answer_inline_query (#2072)

Browse source 
* Auto Pagination

* Fix test_official

* Get things to actually work

* Fine tune

* Tweak tests

* Address review

* Add warning to answer_inline_query
This commit is contained in:
Bibo-Joshi 2020-09-27 14:11:49 +02:00 • committed by GitHub
parent 2989108e95
commit 97adcdf538
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 173 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
telegram/bot.py
View file

@ -39,6 +39,7 @@ from telegram import (User, Message, Update, Chat, ChatMember, UserProfilePhotos
ReplyMarkup, TelegramObject, WebhookInfo, GameHighScore, StickerSet,
PhotoSize, Audio, Document, Sticker, Video, Animation, Voice, VideoNote,
Location, Venue, Contact, InputFile, Poll, BotCommand)
from telegram.constants import MAX_INLINE_QUERY_RESULTS
from telegram.error import InvalidToken, TelegramError
from telegram.utils.helpers import to_timestamp, DEFAULT_NONE
from telegram.utils.request import Request
@ -1526,15 +1527,24 @@ class Bot(TelegramObject):
switch_pm_text=None,
switch_pm_parameter=None,
timeout=None,
current_offset=None,
**kwargs):
"""
Use this method to send answers to an inline query. No more than 50 results per query are
allowed.
Warning:
In most use cases :attr:`current_offset` should not be passed manually. Instead of
calling this method directly, use the shortcut :meth:`telegram.InlineQuery.answer` with
``auto_pagination=True``, which will take care of passing the correct value.
Args:
inline_query_id (:obj:`str`): Unique identifier for the answered query.
results (List[:class:`telegram.InlineQueryResult`)]: A list of results for the inline
query.
results (List[:class:`telegram.InlineQueryResult`] | Callable): A list of results for
the inline query. In case :attr:`current_offset` is passed, ``results`` may also be
a callable accepts the current page index starting from 0. It must return either a
list of :class:`telegram.InlineResult` instances or :obj:`None` if there are no
more results.
cache_time (:obj:`int`, optional): The maximum amount of time in seconds that the
result of the inline query may be cached on the server. Defaults to 300.
is_personal (:obj:`bool`, optional): Pass :obj:`True`, if results may be cached on
@ -1550,6 +1560,10 @@ class Bot(TelegramObject):
switch_pm_parameter (:obj:`str`, optional): Deep-linking parameter for the /start
message sent to the bot when user presses the switch button. 1-64 characters,
only A-Z, a-z, 0-9, _ and - are allowed.
current_offset (:obj:`str`, optional): The :attr:`telegram.InlineQuery.offset` of
the inline query to answer. If passed, PTB will automatically take care of
the pagination for you, i.e. pass the correct ``next_offset`` and truncate the
results list/get the results from the callable you passed.
timeout (:obj:`int` | :obj:`float`, optional): If this value is specified, use it as
the read timeout from the server (instead of the one specified during creation of
the connection pool).
@ -1573,7 +1587,35 @@ class Bot(TelegramObject):
"""
url = '{}/answerInlineQuery'.format(self.base_url)
for res in results:
if current_offset is not None and next_offset is not None:
raise ValueError('`current_offset` and `next_offset` are mutually exclusive!')
if current_offset is not None:
if current_offset == '':
current_offset = 0
else:
current_offset = int(current_offset)
next_offset = ''
if callable(results):
effective_results = results(current_offset)
if not effective_results:
effective_results = []
else:
next_offset = current_offset + 1
else:
if len(results) > (current_offset + 1) * MAX_INLINE_QUERY_RESULTS:
next_offset = current_offset + 1
effective_results = results[
current_offset * MAX_INLINE_QUERY_RESULTS:
next_offset * MAX_INLINE_QUERY_RESULTS]
else:
effective_results = results[current_offset * MAX_INLINE_QUERY_RESULTS:]
else:
effective_results = results
for res in effective_results:
if res._has_parse_mode and res.parse_mode == DEFAULT_NONE:
if self.defaults:
res.parse_mode = self.defaults.parse_mode
@ -1594,10 +1636,8 @@ class Bot(TelegramObject):
else:
res.input_message_content.disable_web_page_preview = None
results = [res.to_dict() for res in results]
data = {'inline_query_id': inline_query_id, 'results': results}
effective_results = [res.to_dict() for res in effective_results]
data = {'inline_query_id': inline_query_id, 'results': effective_results}
if cache_time or cache_time == 0:
data['cache_time'] = cache_time
if is_personal:

25
telegram/inline/inlinequery.py
View file

@ -75,14 +75,21 @@ class InlineQuery(TelegramObject):
return cls(bot=bot, **data)
def answer(self, *args, **kwargs):
def answer(self, *args, auto_pagination=False, **kwargs):
"""Shortcut for::
bot.answer_inline_query(update.inline_query.id, *args, **kwargs)
bot.answer_inline_query(update.inline_query.id,
*args,
current_offset=self.offset if auto_pagination else None,
**kwargs)
Args:
results (List[:class:`telegram.InlineQueryResult`]): A list of results for the inline
query.
results (List[:class:`telegram.InlineQueryResult`] | Callable): A list of results for
the inline query. In case :attr:`auto_pagination` is set to :obj:`True`,
``results`` may also be a callable may also be a callable accepts the current page
index starting from 0. It must return either a list of
:class:`telegram.InlineResult` instances or :obj:`None` if there are no more
results.
cache_time (:obj:`int`, optional): The maximum amount of time in seconds that the
result of the inline query may be cached on the server. Defaults to 300.
is_personal (:obj:`bool`, optional): Pass :obj:`True`, if results may be cached on the
@ -98,6 +105,14 @@ class InlineQuery(TelegramObject):
switch_pm_parameter (:obj:`str`, optional): Deep-linking parameter for the /start
message sent to the bot when user presses the switch button. 1-64 characters,
only A-Z, a-z, 0-9, _ and - are allowed.
auto_pagination (:obj:`bool`, optional): If set to :obj:`True`, :attr:`offset` will be
passed as :attr:`current_offset` to :meth:telegram.Bot.answer_inline_query`.
Defaults to :obj:`False`.
"""
return self.bot.answer_inline_query(self.id, *args, **kwargs)
return self.bot.answer_inline_query(
self.id,
*args,
current_offset=self.offset if auto_pagination else None,
**kwargs
)

95
tests/test_bot.py
View file

@ -27,6 +27,7 @@ from telegram import (Bot, Update, ChatAction, TelegramError, User, InlineKeyboa
InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent,
ShippingOption, LabeledPrice, ChatPermissions, Poll, BotCommand,
InlineQueryResultDocument, Dice, MessageEntity, ParseMode)
from telegram.constants import MAX_INLINE_QUERY_RESULTS
from telegram.error import BadRequest, InvalidToken, NetworkError, RetryAfter
from telegram.utils.helpers import from_timestamp, escape_markdown
from tests.conftest import expect_bad_request
@ -54,6 +55,20 @@ def chat_permissions():
return ChatPermissions(can_send_messages=False, can_change_info=False, can_invite_users=False)
def inline_results_callback(page=None):
if not page:
return [InlineQueryResultArticle(i, str(i), None) for i in range(1, 254)]
elif page <= 5:
return [InlineQueryResultArticle(i, str(i), None)
for i in range(page * 5 + 1, (page + 1) * 5 + 1)]
return None
@pytest.fixture(scope='class')
def inline_results():
return inline_results_callback()
class TestBot:
@pytest.mark.parametrize('token', argvalues=[
'123',
@ -385,6 +400,86 @@ class TestBot:
switch_pm_text='switch pm',
switch_pm_parameter='start_pm')
def test_answer_inline_query_current_offset_error(self, bot, inline_results):
with pytest.raises(ValueError, match=('`current_offset` and `next_offset`')):
bot.answer_inline_query(1234,
results=inline_results,
next_offset=42,
current_offset=51)
@pytest.mark.parametrize('current_offset,num_results,id_offset,expected_next_offset',
[('', MAX_INLINE_QUERY_RESULTS, 1, 1),
(1, MAX_INLINE_QUERY_RESULTS, 51, 2),
(5, 3, 251, '')])
def test_answer_inline_query_current_offset_1(self,
monkeypatch,
bot,
inline_results,
current_offset,
num_results,
id_offset,
expected_next_offset):
# For now just test that our internals pass the correct data
def make_assertion(_, url, data, *args, **kwargs):
results = data['results']
length_matches = len(results) == num_results
ids_match = all([int(res['id']) == id_offset + i for i, res in enumerate(results)])
next_offset_matches = data['next_offset'] == expected_next_offset
return length_matches and ids_match and next_offset_matches
monkeypatch.setattr('telegram.utils.request.Request.post', make_assertion)
assert bot.answer_inline_query(1234, results=inline_results, current_offset=current_offset)
def test_answer_inline_query_current_offset_2(self, monkeypatch, bot, inline_results):
# For now just test that our internals pass the correct data
def make_assertion(_, url, data, *args, **kwargs):
results = data['results']
length_matches = len(results) == MAX_INLINE_QUERY_RESULTS
ids_match = all([int(res['id']) == 1 + i for i, res in enumerate(results)])
next_offset_matches = data['next_offset'] == 1
return length_matches and ids_match and next_offset_matches
monkeypatch.setattr('telegram.utils.request.Request.post', make_assertion)
assert bot.answer_inline_query(1234, results=inline_results, current_offset=0)
inline_results = inline_results[:30]
def make_assertion(_, url, data, *args, **kwargs):
results = data['results']
length_matches = len(results) == 30
ids_match = all([int(res['id']) == 1 + i for i, res in enumerate(results)])
next_offset_matches = data['next_offset'] == ''
return length_matches and ids_match and next_offset_matches
monkeypatch.setattr('telegram.utils.request.Request.post', make_assertion)
assert bot.answer_inline_query(1234, results=inline_results, current_offset=0)
def test_answer_inline_query_current_offset_callback(self, monkeypatch, bot, caplog):
# For now just test that our internals pass the correct data
def test(_, url, data, *args, **kwargs):
results = data['results']
length = len(results) == 5
ids = all([int(res['id']) == 6 + i for i, res in enumerate(results)])
next_offset = data['next_offset'] == 2
return length and ids and next_offset
monkeypatch.setattr('telegram.utils.request.Request.post', test)
assert bot.answer_inline_query(1234, results=inline_results_callback, current_offset=1)
def test(_, url, data, *args, **kwargs):
results = data['results']
length = results == []
next_offset = data['next_offset'] == ''
return length and next_offset
monkeypatch.setattr('telegram.utils.request.Request.post', test)
assert bot.answer_inline_query(1234, results=inline_results_callback, current_offset=6)
@flaky(3, 1)
@pytest.mark.timeout(10)
def test_get_user_profile_photos(self, bot, chat_id):

9
tests/test_inlinequery.py
View file

@ -68,6 +68,15 @@ class TestInlineQuery:
monkeypatch.setattr(inline_query.bot, 'answer_inline_query', test)
assert inline_query.answer()
def test_answer_auto_pagination(self, monkeypatch, inline_query):
def make_assertion(*args, **kwargs):
inline_query_id_matches = args[0] == inline_query.id
offset_matches = kwargs.get('current_offset') == inline_query.offset
return offset_matches and inline_query_id_matches
monkeypatch.setattr(inline_query.bot, 'answer_inline_query', make_assertion)
assert inline_query.answer(auto_pagination=True)
def test_equality(self):
a = InlineQuery(self.id_, User(1, '', False), '', '')
b = InlineQuery(self.id_, User(1, '', False), '', '')

2
tests/test_official.py
View file

@ -78,6 +78,8 @@ def check_method(h4):
ignored |= {'location'} # Added for ease of use
elif name == 'sendVenue':
ignored |= {'venue'} # Added for ease of use
elif name == 'answerInlineQuery':
ignored |= {'current_offset'} # Added for ease of use
assert (sig.parameters.keys() ^ checked) - ignored == set()