Как выполнять функцию через некоторое время в aiogram?

Question

Как сделать что бы функция function выполнялась через некоторое время, например час?

from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor

bot = Bot(token="")
dp = Dispatcher(bot)

@dp.message_handler()
async def message(msg: types.Message):
    ...

async def function(): #эта функция
    ...

if __name__ == "__main__":
    executor.start_polling(dp)

Answer 1

Если триггер - входящее сообщение

используйте asyncio.sleep() или инструменты отложенного вызова

Если нужно запускать функцию по расписанию

Используйте готовый планировщик или напишите свой

Answer 2

Вот пример кода, в котором функция scheduled будет выполнятся каждые 10 секунд (как пример)

import logging
import asyncio
from datetime import datetime

from aiogram import Bot, Dispatcher, executor, types

# Initalialization API token for work with Telegram Bot
API_TOKEN = "BOT TOKEN HERE"

# Configure logging
logging.basicConfig(level=logging.INFO)

# Initialize bot and dispatcher
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler()
async def echo(message: types.Message):
    await message.answer(message.text)

async def scheduled(wait_for):
  while True:
    await asyncio.sleep(wait_for)

    print('Время пришло!')

if __name__ == '__main__':
  loop = asyncio.get_event_loop()
  loop.create_task(scheduled(10)) # поставим 10 секунд, в качестве теста
  executor.start_polling(dp, skip_updates=True)

Удачи!

Answer 3

При использовании asyncio.sleep(), если вдруг бот упадёт, то функция не выполнится. Что бы этого избежать, используйте очереди, например arq.

#scheduler.py
import asyncio
from httpx import AsyncClient
from arq.connections import RedisSettings

async def function(ctx, smth):
    ...


class WorkerSettings:
    functions = [function]

Использование в aiogram

#main.py    
...
from arq import ArqRedis, create_pool
...
arq = await create_pool(RedisSettings())
dp = Dispatcher(bot, arq=arq)
...
@dp.message_handler()
async def message(msg: types.Message, arq: ArqRedis):
    await arq.enqueue_job('function', _defer_by=timedelta(minutes=123), smth=smth)

Помимо запуска main.py, нужно так-же запустить arq из виртуального окружения:

arq scheduler.WorkerSettings

БЛОГ НА HUSL

Как выполнять функцию через некоторое время в aiogram?

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

Если триггер - входящее сообщение

Если нужно запускать функцию по расписанию