Back to top
Edit this page
Toggle table of contents sidebar
This module contains classes that represent Telegram errors.
Changed in version 20.0: Replaced Unauthorized
by Forbidden
.
- exception telegram.error.BadRequest(message)[source]¶
-
Bases:
telegram.error.NetworkError
Raised when Telegram could not process the request correctly.
- exception telegram.error.ChatMigrated(new_chat_id)[source]¶
-
Bases:
telegram.error.TelegramError
Raised when the requested group chat migrated to supergroup and has a new chat id.
- Parameters:
-
new_chat_id (
int
) – The new chat id of the group.
- new_chat_id[source]¶
-
The new chat id of the group.
- Type:
-
int
- exception telegram.error.Conflict(message)[source]¶
-
Bases:
telegram.error.TelegramError
Raised when a long poll or webhook conflicts with another one.
- exception telegram.error.Forbidden(message)[source]¶
-
Bases:
telegram.error.TelegramError
Raised when the bot has not enough rights to perform the requested action.
Changed in version 20.0: This class was previously named
Unauthorized
.
- exception telegram.error.InvalidToken(message=None)[source]¶
-
Bases:
telegram.error.TelegramError
Raised when the token is invalid.
- Parameters:
-
message (
str
, optional) –Any additional information about the exception.
New in version 20.0.
- exception telegram.error.NetworkError(message)[source]¶
-
Bases:
telegram.error.TelegramError
Base class for exceptions due to networking errors.
- exception telegram.error.PassportDecryptionError(message)[source]¶
-
Bases:
telegram.error.TelegramError
Something went wrong with decryption.
Changed in version 20.0: This class was previously named
TelegramDecryptionError
and was available via
telegram.TelegramDecryptionError
.
- exception telegram.error.RetryAfter(retry_after)[source]¶
-
Bases:
telegram.error.TelegramError
Raised when flood limits where exceeded.
Changed in version 20.0:
retry_after
is now an integer to comply with the Bot API.- Parameters:
-
retry_after (
int
) – Time in seconds, after which the bot can retry the request.
- retry_after[source]¶
-
Time in seconds, after which the bot can retry the request.
- Type:
-
int
- exception telegram.error.TelegramError(message)[source]¶
-
Bases:
Exception
Base class for Telegram errors.
- exception telegram.error.TimedOut(message=None)[source]¶
-
Bases:
telegram.error.NetworkError
Raised when a request took too long to finish.
- Parameters:
-
message (
str
, optional) –Any additional information about the exception.
New in version 20.0.
Реализация собственного обработчика ошибок для Telegram бота.
Внимание! Пакеты
python-telegram-bot
версии 13.x будут придерживаться многопоточной парадигмы программирования (на данный момент актуальна версия 13.14). Пакеты версий 20.x и новее будут полностью асинхронными и на октябрь 2022 года, первый из них находится в предрелизе. Дополнительно смотрите основные изменения в пакетеpython-telegram-bot
версии 20.x.
В python-telegram-bot
все ошибки, связанные с Telegram, инкапсулируются в класс исключения TelegramError
и его подклассы, расположенные в модуле telegram.error
.
Любая ошибка, включая TelegramError
, которая возникает в одном из обработчиков сообщений или при вызове Updater.get_updates()
, перенаправляется всем зарегистрированным обработчикам ошибок, чтобы можно было на них отреагировать.
Что бы зарегистрировать обработчик ошибок, необходимо вызвать Dispatcher.add_error_handler(callback)
, где обратный вызов callback
— это функция, которая принимает обновление update
и контекст context
. Объект update
будет обновлением, вызвавшим ошибку (или None
, если ошибка не была вызвана update
, например, для Jobs
), а context.error
— возникшей ошибкой.
: пытаемся отправить сообщение, но пользователь заблокировал бота. Будет вызвано исключение Unauthorized
, подкласса TelegramError
, которое будет доставлено обработчику ошибок. В обработчике ошибок можно удалить этот контакт из списка контактов бота.
. Можно обрабатывать исключения по мере их возникновения. Обработчику ошибок python-telegram-bot
перенаправляются только неперехваченные исключения.
Пример собственного обработчика ошибок.
Очень простой пример того, как можно реализовать собственный обработчик ошибок.
import html import json import logging import traceback from telegram import Update, ParseMode from telegram.ext import Updater, CommandHandler logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) # Токен, который получен от @botfather при создании бота BOT_TOKEN = "TOKEN" # Это может быть ваш собственный идентификатор # или идентификатор группы/канала. # Можно использовать команду `/start` этого бота, # чтобы увидеть свой идентификатор чата. CHAT_ID = 123456789 # Функция-обработчик ошибок def error_handler(update, context): """ Регистрирует ошибку и уведомляет разработчика сообщением telegram. """ # Пишем ошибку, прежде чем что-то делать. Вдруг что-то сломается. logger.error(msg="Исключение при обработке сообщения:", exc_info=context.error) # `traceback.format_exception` возвращает обычное сообщение python # об исключении в виде списка строк, поэтому объединяем их вместе. tb_list = traceback.format_exception(None, context.error, context.error.__traceback__) tb_string = ''.join(tb_list) # Создаем сообщение с некоторой разметкой и дополнительной # информацией о том, что произошло. Возможно, придется добавить некоторую # логику для работы с сообщениями длиной более 4096 символов. update_str = update.to_dict() if isinstance(update, Update) else str(update) message = ( f'Возникло исключение при обработке сообщения.n' f'<pre>update = {html.escape(json.dumps(update_str, indent=2, ensure_ascii=False))}' '</pre>nn' f'<pre>context.chat_data = {html.escape(str(context.chat_data))}</pre>nn' f'<pre>context.user_data = {html.escape(str(context.user_data))}</pre>nn' f'<pre>{html.escape(tb_string)}</pre>' ) # Отправляем сообщение разработчику context.bot.send_message(chat_id=CHAT_ID, text=message, parse_mode=ParseMode.HTML) # объект `update` в функции не используется, заменяем его на `_` def bad_command(_, context): """Вызывает ошибку, чтобы вызвать обработчик ошибок.""" context.bot.wrong_method_name() # объект `context` в функции не используется, заменяем его на `_` def start(update, _): update.effective_message.reply_html( 'Принудительный вызов ошибки `/bad_command`n' f'Ваш идентификатор чата <code>{update.effective_chat.id}</code>.' ) if __name__ == '__main__': updater = Updater(BOT_TOKEN) dispatcher = updater.dispatcher # Зарегистрируем команды... dispatcher.add_handler(CommandHandler('start', start)) dispatcher.add_handler(CommandHandler('bad_command', bad_command)) # ...и обработчик ошибок dispatcher.add_error_handler(error_handler) # Запускаем бота updater.start_polling() updater.idle()
Хороший обработчик ошибок.
Следующий фрагмент кода уведомляет пользователя, когда происходит ошибка, и уведомляет разработчиков об ошибке, включая трассировку и место ее возникновения. Комментарии в коде объясняют, что именно происходит, когда и почему, поэтому редактировать его в соответствии с особыми потребностями должно быть проще простого.
from telegram import ParseMode from telegram.utils.helpers import mention_html import sys import traceback # это общая функция обработчика ошибок. # Если нужна дополнительная информация о конкретном типе сообщения, # добавьте ее в полезную нагрузку в соответствующем предложении `if ...` def error(update, context): # добавьте все идентификаторы разработчиков в этот список. # Можно добавить идентификаторы каналов или групп. devs = [208589966] # Уведомление пользователя об этой проблеме. # Уведомления будут работать, только если сообщение НЕ является # обратным вызовом, встроенным запросом или обновлением опроса. # В случае, если это необходимо, то имейте в виду, что отправка # сообщения может потерпеть неудачу if update.effective_message: text = "К сожалению произошла ошибка в момент обработки сообщения. " "Мы уже работаем над этой проблемой." update.effective_message.reply_text(text) # Трассировка создается из `sys.exc_info`, которая возвращается в # как третье значение возвращаемого кортежа. Затем используется # `traceback.format_tb`, для получения `traceback` в виде строки. trace = "".join(traceback.format_tb(sys.exc_info()[2])) # попробуем получить как можно больше информации из обновления telegram payload = [] # обычно всегда есть пользователь. Если нет, то это # либо канал, либо обновление опроса. if update.effective_user: bad_user = mention_html(update.effective_user.id, update.effective_user.first_name) payload.append(f' с пользователем {bad_user}') # есть ситуаций, когда что то с чатом if update.effective_chat: payload.append(f' внутри чата <i>{update.effective_chat.title}</i>') if update.effective_chat.username: payload.append(f' (@{update.effective_chat.username})') # полезная нагрузка - опрос if update.poll: payload.append(f' с id опроса {update.poll.id}.') # Поместим это в 'хорошо' отформатированный текст text = f"Ошибка <code>{context.error}</code> случилась{''.join(payload)}. " f"Полная трассировка:nn<code>{trace}</code>" # и отправляем все разработчикам for dev_id in devs: context.bot.send_message(dev_id, text, parse_mode=ParseMode.HTML) # Необходимо снова вызывать ошибку, для того, чтобы модуль `logger` ее записал. # Если вы не используете этот модуль, то самое время задуматься. raise
Here are the examples of the python api telegram.TelegramError taken from open source projects. By voting up you can indicate which examples are most useful and appropriate.
3
Example 1
def _check_ssl_cert(self, cert, key):
# Check SSL-Certificate with openssl, if possible
try:
exit_code = subprocess.call(
["openssl", "x509", "-text", "-noout", "-in", cert],
stdout=open(os.devnull, 'wb'),
stderr=subprocess.STDOUT)
except OSError:
exit_code = 0
if exit_code is 0:
try:
self.httpd.socket = ssl.wrap_socket(
self.httpd.socket, certfile=cert, keyfile=key, server_side=True)
except ssl.SSLError as error:
self.logger.exception('Failed to init SSL socket')
raise TelegramError(str(error))
else:
raise TelegramError('SSL Certificate invalid')
3
Example 2
@staticmethod
def is_image(stream):
"""Check if the content file is an image by analyzing its headers.
Args:
stream (str): A str representing the content of a file.
Returns:
str: The str mimetype of an image.
"""
image = imghdr.what(None, stream)
if image:
return 'image/%s' % image
raise TelegramError('Could not parse file content')
3
Example 3
@flaky(3, 1)
@timeout(10)
def test_error_send_audio_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['audio'] = open(os.devnull, 'rb')
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendAudio(chat_id=self._chat_id, **json_dict))
3
Example 4
@flaky(3, 1)
@timeout(10)
def test_error_send_audio_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['audio'] = ''
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendAudio(chat_id=self._chat_id, **json_dict))
3
Example 5
def testInvalidSrvResp(self):
with self.assertRaisesRegexp(telegram.TelegramError, 'Invalid server response'):
# bypass the valid token check
newbot_cls = type(
'NoTokenValidateBot', (telegram.Bot,), dict(_validate_token=lambda x, y: None))
bot = newbot_cls('0xdeadbeef')
bot.base_url = 'https://api.telegram.org/bot{0}'.format('12')
bot.getMe()
3
Example 6
@flaky(3, 1)
@timeout(10)
def test_error_send_docuement_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['docuement'] = open(os.devnull, 'rb')
self.assertRaises(telegram.TelegramError,
lambda: self._bot.sendDocuement(chat_id=self._chat_id,
**json_dict))
3
Example 7
@flaky(3, 1)
@timeout(10)
def test_error_send_docuement_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['docuement'] = ''
self.assertRaises(telegram.TelegramError,
lambda: self._bot.sendDocuement(chat_id=self._chat_id,
**json_dict))
3
Example 8
def test_error_get_empty_file_id(self):
json_dict = self.json_dict
json_dict['file_id'] = ''
del (json_dict['file_path'])
del (json_dict['file_size'])
self.assertRaises(telegram.TelegramError, lambda: self._bot.getFile(**json_dict))
3
Example 9
def test_error_send_location_empty_args(self):
json_dict = self.json_dict
json_dict['latitude'] = ''
json_dict['longitude'] = ''
self.assertRaises(telegram.TelegramError,
lambda: self._bot.sendLocation(chat_id=self._chat_id,
**json_dict))
3
Example 10
@flaky(3, 1)
@timeout(10)
def test_error_send_photo_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['photo'] = open(os.devnull, 'rb')
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendPhoto(chat_id=self._chat_id, **json_dict))
3
Example 11
@flaky(3, 1)
@timeout(10)
def test_error_send_photo_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['photo'] = ''
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendPhoto(chat_id=self._chat_id, **json_dict))
3
Example 12
@flaky(3, 1)
@timeout(10)
def test_error_send_sticker_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['sticker'] = open(os.devnull, 'rb')
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendSticker(chat_id=self._chat_id, **json_dict))
3
Example 13
@flaky(3, 1)
@timeout(10)
def test_error_send_sticker_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['sticker'] = ''
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendSticker(chat_id=self._chat_id, **json_dict))
3
Example 14
def test_addRemoveErrorHandler(self):
self._setup_updater('', messages=0)
d = self.updater.dispatcher
d.add_error_handler(self.errorHandlerTest)
queue = self.updater.start_polling(0.01)
error = TelegramError("Unauthorized.")
queue.put(error)
sleep(.1)
self.assertEqual(self.received_message, "Unauthorized.")
# Remove handler
d.remove_error_handler(self.errorHandlerTest)
self.reset()
queue.put(error)
sleep(.1)
self.assertTrue(None is self.received_message)
3
Example 15
def getUpdates(self, offset=None, limit=100, timeout=0, network_delay=2.):
if self.raise_error:
raise TelegramError('Test Error 2')
elif self.send_messages >= 2:
self.send_messages -= 2
return self.mockUpdate(self.text), self.mockUpdate(self.text)
elif self.send_messages == 1:
self.send_messages -= 1
return self.mockUpdate(self.text),
else:
return []
3
Example 16
@flaky(3, 1)
@timeout(10)
def test_error_send_video_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['video'] = open(os.devnull, 'rb')
self.assertRaises(telegram.TelegramError,
lambda: self._bot.sendVideo(chat_id=self._chat_id,
timeout=10,
**json_dict))
3
Example 17
@flaky(3, 1)
@timeout(10)
def test_error_send_video_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['video'] = ''
self.assertRaises(telegram.TelegramError,
lambda: self._bot.sendVideo(chat_id=self._chat_id,
timeout=10,
**json_dict))
3
Example 18
@flaky(3, 1)
@timeout(10)
def test_error_send_voice_empty_file(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['voice'] = open(os.devnull, 'rb')
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendVoice(chat_id=self._chat_id, **json_dict))
3
Example 19
@flaky(3, 1)
@timeout(10)
def test_error_send_voice_empty_file_id(self):
json_dict = self.json_dict
del (json_dict['file_id'])
json_dict['voice'] = ''
self.assertRaises(
telegram.TelegramError,
lambda: self._bot.sendVoice(chat_id=self._chat_id, **json_dict))
0
Example 20
def start(self):
"""
Thread target of thread 'dispatcher'. Runs in background and processes
the update queue.
"""
if self.running:
self.logger.warning('already running')
return
if self.__exception_event.is_set():
msg = 'reusing dispatcher after exception event is forbidden'
self.logger.error(msg)
raise TelegramError(msg)
self._init_async_threads(uuid4(), self.workers)
self.running = True
self.logger.debug('Dispatcher started')
while 1:
try:
# Pop update from update queue.
update = self.update_queue.get(True, 1)
except Empty:
if self.__stop_event.is_set():
self.logger.debug('orderly stopping')
break
elif self.__exception_event.is_set():
self.logger.critical('stopping due to exception in another thread')
break
continue
self.logger.debug('Processing Update: %s' % update)
self.process_update(update)
self.running = False
self.logger.debug('Dispatcher thread stopped')
0
Example 21
def __init__(self, data):
self.data = data
self.boundary = choose_boundary()
if 'audio' in data:
self.input_name = 'audio'
self.input_file = data.pop('audio')
elif 'docuement' in data:
self.input_name = 'docuement'
self.input_file = data.pop('docuement')
elif 'photo' in data:
self.input_name = 'photo'
self.input_file = data.pop('photo')
elif 'sticker' in data:
self.input_name = 'sticker'
self.input_file = data.pop('sticker')
elif 'video' in data:
self.input_name = 'video'
self.input_file = data.pop('video')
elif 'voice' in data:
self.input_name = 'voice'
self.input_file = data.pop('voice')
elif 'certificate' in data:
self.input_name = 'certificate'
self.input_file = data.pop('certificate')
else:
raise TelegramError('Unknown inputfile type')
if hasattr(self.input_file, 'read'):
self.filename = None
self.input_file_content = self.input_file.read()
if 'filename' in data:
self.filename = self.data.pop('filename')
elif hasattr(self.input_file, 'name'):
# on py2.7, pylint fails to understand this properly
# pylint: disable=E1101
self.filename = os.path.basename(self.input_file.name)
try:
self.mimetype = self.is_image(self.input_file_content)
if not self.filename or '.' not in self.filename:
self.filename = self.mimetype.replace('/', '.')
except TelegramError:
self.mimetype = mimetypes.guess_type(self.filename)[0] or DEFAULT_MIME_TYPE
0
Example 22
@staticmethod
def _parse(json_data):
"""Try and parse the JSON returned from Telegram.
Returns:
dict: A JSON parsed as Python dict with results - on error this dict will be empty.
"""
decoded_s = json_data.decode('utf-8')
try:
data = json.loads(decoded_s)
except ValueError:
raise TelegramError('Invalid server response')
if not data.get('ok'):
description = data.get('description')
parameters = data.get('parameters')
if parameters:
migrate_to_chat_id = parameters.get('migrate_to_chat_id')
if migrate_to_chat_id:
raise ChatMigrated(migrate_to_chat_id)
retry_after = parameters.get('retry_after')
if retry_after:
raise RetryAfter(retry_after)
if description:
return description
return data['result']
0
Example 23
def errorRaisingHandlerTest(self, bot, update):
raise TelegramError(update)
0
Example 24
def handle(self, upd):
if upd.message is not None:
self.logger.debug("Got message: " + str(upd.message))
text = upd.message.text
action = None
if len(text)is not 0:
if text[:1] == '/' and len(text) > 1:
command, *args = text[1:].split()
if '@' in command:
command, uname, *_ = command.lower().split('@')
calling_me = (uname == self.tg.username.lower())
if '@' not in command or calling_me:
action = self.handle_cmd, upd.message, command.lower(), args
else:
action = self.handle_chat, upd.message
else:
action = self.handle_other, upd.message
if action is not None:
fn, *args = action
try:
fn(*args)
except telegram.TelegramError as e:
self.logger.warning(
"TelegramError while handling message: " + str(e))
self.ack(upd)