Узнайте, как использовать Middleware в ботах Telegram с помощью Aiogram и превратите свой бот в более удобный и функциональный инструмент. Middleware — это мощный инструмент для добавления кастомных данных в хэндлер или остановки обработки апдейта. Посмотрите наше видео, чтобы узнать больше об этой уникальной функции Aiogram.
Исходный код к уроку
handlers/basic.py
from aiogram import Bot
from aiogram.types import Message
from core.keyboards.reply import get_reply_keyboard
from core.utils.dbconnect import Request
async def get_start(message: Message, bot: Bot, counter: str, request: Request):
await request.add_data(message.from_user.id, message.from_user.first_name)
await message.answer(f'Сообщение #{counter}')
await message.answer(f'<s>Привет {message.from_user.first_name}. Рад тебя видеть!</s>',
reply_markup=get_reply_keyboard())midlleware/countermiddleware.py
from aiogram import BaseMiddleware
from aiogram.types import Message
from typing import Dict, Any, Callable, Awaitable
class CounterMiddleware(BaseMiddleware):
def __init__(self) -> None:
self.counter = 0
async def __call__(
self,
handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
event: Message,
data: Dict[str, Any]
) -> Any:
self.counter += 1
data['counter'] = self.counter
return await handler(event, data)midlleware/dbmiddleware.py
from typing import Callable, Awaitable, Dict, Any
from psycopg_pool import AsyncConnectionPool
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
from core.utils.dbconnect import Request
class DbSession(BaseMiddleware):
def __init__(self, connector: AsyncConnectionPool):
super().__init__()
self.connector = connector
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
async with self.connector.connection() as connect:
data['request'] = Request(connect)
return await handler(event, data)midlleware/officehours.py
from datetime import datetime
from aiogram import BaseMiddleware
from typing import Callable, Awaitable, Dict, Any
from aiogram.types import Message, TelegramObject
def office_hours() -> bool:
return datetime.now().weekday() in (0, 1, 2, 3, 4) and datetime.now().hour in ([i for i in (range(8, 19))])
class OfficeHoursMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
if office_hours():
return await handler(event, data)
# await event.answer('Время работы бота:\r\nПн-пт с 8 до 18. Приходите в рабочие часы.')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
import psycopg_pool
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()
dp.update.middleware.register(DbSession(pool_connect))
dp.message.middleware.register(CounterMiddleware())
dp.update.middleware.register(OfficeHoursMiddleware())
dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)
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())
