Как отлавливать TimeoutError в asyncio

Question

Подскажите пожалуйста как отлавливать TimeoutError. Я запускаю телеграмм бота он работает все в порядке, потом при потере интернет соединения asyncio.exceptions.TimeoutError.

Вопрос в следующем, как избежать данной ошибки? То есть я хочу чтобы после того как соединение вновь появлялось, бот запускался самостоятельно после восстановления соединения.

Код:

from aiogram import Bot
from aiogram.dispatcher import Dispatcher
from aiogram.contrib.fsm_storage.memory import MemoryStorage

storage = MemoryStorage()

TOKEN = open('token.txt', 'r').readline()    
bot = Bot(token=TOKEN)
dp = Dispatcher(bot, storage=storage)

async def on_startup(_):
    print('Бот запущен!')


def main():
    executor.start_polling(dp, skip_updates=True, on_startup=on_startup)



if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print('Произошла ошибка!')

Answer 1

Вешаете декоратор @TimeoutErrorHandler над проблемной функцией и наслаждаетесь логами о перехваченной ошибке. Этот декоратор просто вызывает функцию, над которой его повесели, в бесконечном цикле, при TimeoutError он ждет 1 секунду и снова запускает функцию. Если вдруг вылезет другая ошибка, то ей ничего не помешает.

from asyncio.exceptions import TimeoutError
from time import sleep

def TimeoutErrorHandler(func):
    def wrap(*args, **kwargs):
        while True:
            try:
                func(*args, **kwargs)
            except TimeoutError:
                print(f'TimeoutError excepted, wait 1 second')
                sleep(1)
    return wrap

@TimeoutErrorHandler
def func_with_error():
    raise TimeoutError

func_with_error()

Логи будут выглядеть примерно так:

TimeoutError excepted, wait 1 second
TimeoutError excepted, wait 1 second
...

БЛОГ НА HUSL

Как отлавливать TimeoutError в asyncio

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