Как реализовать отправку уведомлений ботом aiogram?

Question

[Martin_Shelby]

Пишу tg бота на базе aiogram. На данный момент у меня запускается бот и если приходит какое-то сообщение от пользователя - оно обрабатывается и выдаёт результат. Если никто не пишет, то соответственно он просто ждёт обновления и ничего не делает. А я хочу, чтобы он отправлял уведомление не в результате каких-то действий пользователя бота, а по некоторым событиям происходящим на сервере, например истёкшее время на ответ и тд. Как это можно сделать?

Пример: бот отправил запрос клиенту и у него есть 30 минут на обработку этого запроса, если в течении этого времени клиент не ответил на запрос, то ему нужно отправить уведомление об истечении запроса.

Comments

[shurshur]

Запустить отдельную функцию, которая будет с интервалом типа await asyncio.sleep(60) проверять не нужно ли кому-то ответить и отвечать?

[MinTnt]

Создать базу данных, в которую сохранять какому ID'шнику, во сколько отправить какой-то текст

[Martin_Shelby]

shurshur, вы это имеете в виду?

async def func():
    body
    await asyncio.sleep(60)
asyncio.run(func())

[Martin_Shelby]

MinTnt, нет проблем с реализацией задачи, есть проблема в отправке уведомления.

[shurshur]

Martin_Shelby, примерно так:

async def background_task():
  while True:
    if some_messages_ready_to_send:
      send_messages
    await asyncio.sleep(60)
asyncio.create_task(background_task())

main_script_code...

[Martin_Shelby]

shurshur, я уже пробовал сделать что-то подобное, в результате получаю это:

Traceback (most recent call last):
  File "F:\Projects\Git\tg_bot\app.py", line 20, in <module>
    asyncio.create_task(background_task())
  File "C:\Users\Mearise\AppData\Local\Programs\Python\Python39\lib\asyncio\tasks.py", line 360, in create_task
    loop = events.get_running_loop()
RuntimeError: no running event loop
sys:1: RuntimeWarning: coroutine 'background_task' was never awaited

[shurshur]

Martin_Shelby, вот тут пример как такое решить: https://stackoverflow.com/questions/58774718/async...

Вкратце: либо надо сделать:

loop = asyncio.get_event_loop()
loop.create_task(background_task())

(инициализировать event loop и уже в нём создать task), либо вынести основной код программы в отдельную функцию и её вызвать с помощью run_until_complete:

async def main():
  asyncio.create_task(background_task())
  do_somethin_other

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

[Martin_Shelby]

shurshur, первый вариант сработал. Благодарю

[TaPX]

Martin_Shelby, а вот это писать до или после поллинга? или без разницы?