Telegram-Bot in project with different threads

[feller1002]

I have a relatively complex program which processes data from different interfaces. For this purpose, each interface runs in its own thread. The threads exchange data via queues. When certain events occur, I am notified via Telegram. I now need to update the Python version (v3.8 -> v3.11) and therefore also the Telegram library (v8 ->v20.5).

For example:
Thread1 is handling a COM port and puts a message to the queue. The telegram thread (and maybe other threads) is processing the message and sends a message to users.
Thread2 is handling a camera and saves photos. The telegram thread is getting the message with the location of the photo and sends this photo to users.
etc.

I am now having problems with the new implementation with async functions. I have already read that you should not use async functions and threading in parallel. I don't want to rewrite all my code now. Therefore I need an approach how to solve the problem.

[Bibo-Joshi]

Hi. I guess a sane way to get this working is

• Keep all PTB Code in one thread
• Stick to using classic non-asyncio queues
• Use asyncio.loop.run_in_executor to call queue.get in an asyncio friendly way

[feller1002]

Do you have a short working example? I have some trouble with the implementation

[Bibo-Joshi]

No, I don't have anything at hand. Please specify which part is causing you troubles

[feller1002]

I tried to make a short working example.
Problem with this example:
The thread handling the queues does not start.

import threading
import logging
import asyncio
import time
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, BaseHandler, ContextTypes, MessageHandler, filters

class TelegramBotHandler():
	def __init__(self):
		self.application = Application.builder().token(TOKEN).build()
		self.application.add_handler(CommandHandler("help", self.help_command))

		self.application.run_polling(allowed_updates=Update.ALL_TYPES)

	async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
		await update.message.reply_text("Help!")

	def queuehandlerwrapper(self):
		asyncio.loop.run_in_executor(self.queuehandler())

	def sendMSG(self):
		self.application.bot.send_message(MSG_ID, "Hello World")

	async def queuehandler(self):
		print("Queuehandler started")
		while True:
			time.sleep(5)
			# Here I regularly process other thread data from the queue
			await self.sendMSG()

if __name__ == "__main__":
	telebot = TelegramBotHandler()

	queuehandlerthread = threading.Thread(target=telebot.queuehandler, args=())
	queuehandlerthread.start()

[Bibo-Joshi]

Here you try to use the coroutine method queuehandler as target for a thread. that doesn't work.
Have a look at the following MWE that I cooked up. It's certainly not perfect and needs several improvements before being put into production, but I guess it get's a general idea across:

import asyncio
import threading
import time

from telegram.ext import Application, MessageHandler, filters, TypeHandler

# asyncio queue to be used by the PTB application
update_queue = asyncio.Queue()
# event to stop the thread if needed
stop_event = threading.Event()


def thread_target():
    print("Thread started")
    # create a new event loop for this thread
    loop = asyncio.new_event_loop()

    count = 0
    while not stop_event.is_set():
        count += 1
        # put updates from somewhere to the queue
        print(f"Putting update {count} to queue")
        loop.run_until_complete(update_queue.put(count))
        time.sleep(5)


async def process_update(update, context):
    print(f"Bot got update: {update} of type {type(update)}")


async def echo_message(update, context):
    await update.effective_message.reply_text(update.effective_message.text)


def main():
    application = Application.builder().token("TOKEN").update_queue(update_queue).build()

    # register handlers
    application.add_handler(MessageHandler(filters.TEXT, echo_message))
    application.add_handler(TypeHandler(int, process_update))

    # start the thread
    thread = threading.Thread(target=thread_target)
    thread.start()

    # start the application
    application.run_polling()

    # stop the thread
    # this line is reached when the application is stopped
    stop_event.set()
    thread.join()


if __name__ == "__main__":
    main()

[AleNovelli]

Hi, I also have a similar program to yours (many classes to manage com ports and devices using threads) and I was trying to add a telegram bot to notify me of given events. However, I am finding some difficulties in implementing the bot in an elegant way; could you share the code?
It might give me some ideas on the best implementation.

[feller1002]

My program looks something like this:

import threading
import logging
import asyncio
import time
import queue
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
import Thread1
import Thread2

telegram_queue = Queue.queue

class TelegramBotHandler():
    def __init__(self, telegram_queue):
        self.application = Application.builder().token(TOKEN).build()
        self.application.add_handler(CommandHandler("help", self.help_command))
        self.queue = telegram_queue

    def runPolling(self):
        self.application.run_polling(allowed_updates=Update.ALL_TYPES)

    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        await update.message.reply_text("Help!")

    def queuehandler(self):
        while True:
            time.sleep(60)
            # Here I regularly process other thread data from the queue 
            self.application.bot.send_message(MSG_ID, "Hello World")

if __name__ == "__main__":
    # Init of COM Thread
    Thread1 = threading.thread(COMHandler, args=(telegram_queue))
    Thread1.start()

    # Init of Camera Thread
    Thread2 = threading.thread(CameraHandler, args=(telegram_queue))
    Thread2.start()

    telebot = TelegramBotHandler(telegram_queue)
    telebotthread = threading.Thread(target=telebot.runPolling, args=())
    telebotthread.start()

    queuehandlerthread = threading.Thread(target=telebot.queuehandler, args=())
    queuehandlerthread.start()
`

[AleNovelli]

Thanks

[feller1002]

Do you already have a working example ?