How can I break out of telegram bot loop application.run_polling()?

Question

def bot_start():
    application = ApplicationBuilder().token("api_key").build()


    async def stop(update, context):
        await context.bot.send_message(chat_id=update.message.chat_id, text='Terminating Bot...')
        await application.stop()
        await Updater.shutdown(application.bot)
        await application.shutdown()

    async def error(update, context):
        err = f"Update: {update}\nError: {context.error}"
        logging.error(err, exc_info=context.error)
        await context.bot.send_message(chat_id=user_id, text=err)

   application.add_handler(CommandHandler('stop', stop))
   application.add_error_handler(error)

   application.run_polling()

I tried everything I could to stop it and I couldnt as it's not letting other lines of code that comes after calling bot_start() run. It basically never reaches them.

Ombudsman

Answer 1

Application.run_polling is a convenience methods that starts everything and keeps the bot running until you signal the process to shut down. It's mainly intended to be used if the Application is the only long-running thing in your python process. If you want to run other things alongside your bot, you can instead manually call the methods listed in the docs of run_polling. You may also want to have a look at this example, where this is showcased for a setup for a custom webhook server is used instead of PTBs built-in one.

---

Disclaimer: I'm currently the maintainer of python-telegram-bot.

Azilian

Umm I'm kinda still stuck on how can i signal the run_polling to stop – Ombudsman

Answer 2

After searching through the documentation for version 20 of python-telegram-bot, I found out that using the telegram.ext.Updater class with an asyncio.Queue is the key to getting the telegram updates without blocking the main thread. The telegram application, the updater, and the queue should be initialized, and then one can use either que.get() asynchronously to wait for the next update or que.get_nowait() to get a new update instantaneously, which raises an exception if non are available.

import os
import asyncio
import telegram.ext

TELEGRAM_BOT_TOKEN = "ADD YOUR TOKEN HERE"

async def main():
    que = asyncio.Queue()
    application = telegram.ext.ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()
    updater = telegram.ext.Updater(application.bot, update_queue=que)
    await updater.initialize()
    await updater.start_polling()
    
    # Option 1
    update = await que.get()
    
    # Option 2, replace option 1 if you like
    # update = que.get_nowait()
    
    print(update)

asyncio.run(main())

Jehiel

Answer 3

You can call context.application.stop_running() in order to inform the run_polling() method to stop. Documentation link

async def admin_exit(update: Update, context: CallbackContext) -> None:
    ...
    context.application.stop_running()

Reyreyes