'Best practice in sending message in python telegram bot from a 3d party listener
I have a custom code that does its routine and I want to send a message to myself in Telegram if something goes wrong. In my case I use python-telegram-bot
library along with apscheduler
and its listeners, where certain events could be catched.
I came up with such working code, but my question is: is it possible to make it better, namely without using global variable? This was done to overcome the problem that listeners do not accept arguments needed for bot to send a message.
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from telegram.ext import Updater, CommandHandler
import copy
import my_custom_library
saved_update = None
def my_listener(event): # not related with bot
if event.exception:
if saved_update is not None:
alert(saved_update, 'Scheduler threw event.exception.') # should have bot related args
else:
record = event.retval # get returned value from my_custom_library.repetitive_function
try:
processed_record = my_custom_library.my_unsafe_business_logic(record) # something might go wrong here
my_custom_library.add_to_db(processed_record) # and here
except Exception as e:
if saved_update is not None:
alert(saved_update, e) # should have bot related args
def start(update, context):
global saved_update
saved_update = copy.deepcopy(update) # this is what I don't like
update.message.reply_text('You have subscribed for notifications.')
def alert(update, reason):
update.message.reply_text('Something went wrong: {}'.format(reason))
def main():
scheduler = BackgroundScheduler()
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
scheduler.add_job(my_custom_library.repetitive_function, args=(my_args,), trigger='interval', minutes=1)
scheduler.start()
# bot
updater = Updater(TOKEN, use_context=True)
dp = updater.dispatcher
dp.add_handler(CommandHandler("start", callback=start))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
Solution 1:[1]
The Telegram Bot API is fairly simple, you just ned to send an HTTP GET Request to this URL: https://api.telegram.org/bot_token_/sendMessage?chat_id=123&text=Hello%20World!
Just create a bot with Botfather and send the Bot a message. With the specified Token from Botfather and this URL: https://api.telegram.org/bot_token_/getUpdates
You can get the messages which were sent to the Bot and the chat_id.
The simplest way would be to use the requests module and send the output of the updater to the first URL as the text parameter.
Solution 2:[2]
In my scripts I often use callbacks. It's a clean solution that provides separation between the two scripts. The send_message function can accept kwargs, so you can essentially create a dictionary, and update it when sending the message. The first part (channel ID) is something you know on the bot-side, and the second part (the text itself) is something you know one the third-party side.
In your third-party, provide a set_cb function that accepts a callback and a dictionary. Like so:
def set_cb(self, callback, params):
self.callback = callback
self.callback_params = params
In your bot script, set the callback before updater.idle()
# Start the Bot
updater = Updater("TOKEN")
updater.start_polling()
# Set the callback
r.set_cb(updater.bot.send_message, {"chat_id": 123456})
Then, in your third-party, once you want to send a message, simply add the message text call the following:
self.callback_params.update({"text": text})
self.callback(**self.callback_params) # converts the dict to kwargs
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
Solution | Source |
---|---|
Solution 1 | Matthias Schaffer |
Solution 2 | Alex Osheter |