Python3 проблема с threading и asyncio

Question

Очень не хочу лезть в потоково-асинхронную кашу, но я уже в ней.

Написал скрипт, который делает бэкапы сервера и отправляет отчеты о них.
Помимо этого, он отчитывается о статусе сервера и статусе бэкапов в телеграм боте через библиотеку aiogram.

То-есть, разделите мой код на 2 части:

Бекапер
Бот

Проблема:

Я пытаюсь запустить процесс создания бэкапов в отдельном потоке.
Из этого же потока я хочу отправлять сообщения от имени бота подписчикам.

Создание потока под бэкапы

Thread(target=BackUp().start).start()

Функция рассылки подписчикам

async def notify_staff(msg: str, kind='Info', notification=True) -> None:
    staff = [x for x in Session().query(Staff)]
    message = '*\\[{kind}\\]*\n{msg}'.format(kind=markdown_decoration.quote(kind),
                                           msg=markdown_decoration.quote(msg))
    for obj in staff:
        await bot.send_message(chat_id=obj.chat_id, text=message, parse_mode='MarkdownV2',
                               disable_notification=not notification)
        sleep(0.4)

И суть проблемы в том, что вызов

class BackUp:
    def start(self):
        loop = asyncio.new_event_loop()
        while True:
            ...
            asyncio.run_coroutine_threadsafe(notify_staff(msg='Backup done\nStored paths:\n{paths}'.format(
                        paths='\n'.join([x['path'] for x in stored])
                    ), notification=False), loop)
            ...

выполняется бесконечно. (Дебаг помог)
Точнее, он, похоже, просто бесконечно ничего не делает.

Была попытка отправить в другой поток непосредственно бота и все заработало... за исключением бота, который уже не откликался на команды.

Не понимаю почему не отрабатывает run_coroutine_threadsafe()

Answer 1

Проблема в том что ты мешаешь код в разных тредах. Тред бакупа не должен знать ничего о телеграмме и тем более не должен создавать новый луп. Используй concurrent.futures.ThreadPoolExecutor

executor = concurrent.futures.ThreadPoolExecutor()
backup_future = loop.run_in_executor(executor, BackUp().start)
wait_complete_task = loop.create_task(on_backup_done(backup_future))

где в on_backup_done описывается взаимодействеие с телеграмом, запущенном в основном лупе и выполняется оно в основном треде.

async def on_backup_done(backup_future):
    result = await backup_future
    await notify_staff(msg='Backup done....)

БЛОГ НА HUSL

Python3 проблема с threading и asyncio

Проблема:

Ответы (1 шт):