Если вам нужно запускать функции в определенное время или через определенный промежуток, то вы можете использовать apscheduler в aiogram. В этом видео мы покажем, как можно пробросить apscheduler через middleware и использовать его для автоматического запуска функций в вашем боте. Не пропустите этот увлекательный урок и узнайте, как эффективно использовать apscheduler в aiogram.
Исходный код к уроку
handlers/apsched.py
from aiogram import Bot
async def send_message_time(bot: Bot):
await bot.send_message(5528605206, f'Это сообщение отправлено через несколько секунд после старта бота')
async def send_message_cron(bot: Bot):
await bot.send_message(5528605206, f'Это сообщение будет отправляться ежедневно в указанное время.')
async def send_message_interval(bot: Bot):
await bot.send_message(5528605206, f'Это сообщение будет отправляться с интервалом в 1 минуту')
async def send_message_middleware(bot: Bot, chat_id: int):
await bot.send_message(chat_id, f'Это сообщение отправлено с помощью сформированной через middleware задачи')
handlers/form.py
from aiogram.types import Message
from aiogram.fsm.context import FSMContext
from core.utils.statesform import StepsForm
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from core.handlers.apsched import send_message_middleware
from datetime import datetime, timedelta
from aiogram import Bot
async def get_age(message: Message, bot: Bot, state: FSMContext, apscheduler: AsyncIOScheduler):
await message.answer(f'Твой возраст:\r\n{message.text}\r\n')
context_data = await state.get_data()
await message.answer(f'Сохраненные данные в машине состояний:\r\n{str(context_data)}')
name = context_data.get('name')
last_name = context_data.get('last_name')
data_user = f'Вот твои данные\r\n' \
f'Имя {name}\r\n' \
f'Фамилия {last_name}\r\n' \
f'Возраст {message.text}'
await message.answer(data_user)
await state.clear()
apscheduler.add_job(send_message_middleware, trigger='date', run_date=datetime.now() + timedelta(seconds=10),
kwargs={'bot': bot, 'chat_id': message.from_user.id})main.py
from aiogram import Bot, Dispatcher
from aiogram.types import Message, ContentType
from core.handlers.basic import get_start, get_photo, get_hello
from core.filters.iscontact import IsTrueContact
from core.handlers.contact import get_fake_contact, get_true_contact
import asyncio
import logging
from core.settings import settings
from aiogram.filters import ContentTypesFilter, Command, CommandStart
from aiogram import F
from core.utils.commands import set_commands
from core.handlers.basic import get_location
from core.handlers.basic import get_inline
from core.handlers.callback import select_macbook
from core.utils.callbackdata import MacInfo
from core.handlers.pay import order, pre_checkout_query, successful_payment, shipping_check
from core.middlewares.countermiddleware import CounterMiddleware
from core.middlewares.officehours import OfficeHoursMiddleware
from core.middlewares.dbmiddleware import DbSession
from core.middlewares.apschedulermiddleware import SchedulerMiddleware
import psycopg_pool
from core.handlers import form
from core.utils.statesform import StepsForm
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from core.handlers import apsched
from datetime import datetime, timedelta
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
async def start_bot(bot: Bot):
await set_commands(bot)
await bot.send_message(settings.bots.admin_id, text='Бот запущен!')
async def stop_bot(bot: Bot):
await bot.send_message(settings.bots.admin_id, text='Бот остановлен!')
def create_pool():
return psycopg_pool.AsyncConnectionPool(f"host=127.0.0.1 port=5432 dbname=users user=postgres password=qwerty "
f"connect_timeout=60")
async def start():
logging.basicConfig(level=logging.INFO,
format="%(asctime)s - [%(levelname)s] - %(name)s - "
"(%(filename)s).%(funcName)s(%(lineno)d) - %(message)s"
)
bot = Bot(token=settings.bots.bot_token, parse_mode='HTML')
pool_connect = create_pool()
dp = Dispatcher()
scheduler = AsyncIOScheduler(timezone="Europe/Moscow")
scheduler.add_job(apsched.send_message_time, trigger='date', run_date=datetime.now() + timedelta(seconds=10),
kwargs={'bot': bot})
scheduler.add_job(apsched.send_message_cron, trigger='cron', hour=datetime.now().hour,
minute=datetime.now().minute + 1, start_date=datetime.now(), kwargs={'bot': bot})
scheduler.add_job(apsched.send_message_interval, trigger='interval', seconds=60, kwargs={'bot': bot})
scheduler.start()
dp.update.middleware.register(DbSession(pool_connect))
dp.message.middleware.register(CounterMiddleware())
dp.update.middleware.register(OfficeHoursMiddleware())
dp.update.middleware.register(SchedulerMiddleware(scheduler))
dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)
dp.message.register(form.get_form, Command(commands='form'))
dp.message.register(form.get_name, StepsForm.GET_NAME)
dp.message.register(form.get_last_name, StepsForm.GET_LAST_NAME)
dp.message.register(form.get_age, StepsForm.GET_AGE)
dp.message.register(order, Command(commands='pay'))
dp.pre_checkout_query.register(pre_checkout_query)
dp.message.register(successful_payment, ContentTypesFilter(content_types=[ContentType.SUCCESSFUL_PAYMENT]))
dp.shipping_query.register(shipping_check)
dp.message.register(get_inline, Command(commands='inline'))
dp.callback_query.register(select_macbook, MacInfo.filter(F.model == 'pro'))
# dp.message.register(get_photo, ContentTypesFilter(content_types=[ContentType.PHOTO]))
dp.message.register(get_location, ContentTypesFilter(content_types=[ContentType.LOCATION]))
dp.message.register(get_hello, F.text == 'Привет')
dp.message.register(get_true_contact, ContentTypesFilter(content_types=[ContentType.CONTACT]), IsTrueContact())
dp.message.register(get_fake_contact, ContentTypesFilter(content_types=[ContentType.CONTACT]))
dp.message.register(get_photo, F.photo)
dp.message.register(get_start, Command(commands=['start', 'run']))
# dp.message.register(get_start, CommandStart)
try:
await dp.start_polling(bot)
finally:
await bot.session.close()
if __name__ == "__main__":
asyncio.run(start())
