Гайд Aiogram 2 - Middlewares

[SCOTE]

Сейчас я вам расскажу про такой инструмент разработки как мидлвари в Aiogram 2.

Зачем они нужны? Конкретно в аиограме используется как прослойка между библиотекой и вашим кодом.

async def welcome(message: Message, repo: Repo, state: FSMContext):
    if not await repo.get_user(message.from_user.id):
        await repo.add_user(message.from_user.id)
        

def register_command_start(dp: Dispatcher):
    dp.register_message_handler(welcome, commands=["start"], state='*')

Что тут проиходит: при команде /start мы напрямую обращаемся из хендлера к БД (repo) и записываем айдишник нового пользователя, никакие глобальные переменные не нужны. И, между прочем, к БД мы можем обратиться из любого хендлера, что добавляет некие удобства и организованность кода.

Перейдем к сути, сейчас покажу как зарегать мидлварь в коде на простом примере. Будем записывать в json-файл ID всех пользователей, которые зайдут в бота. (Я не хочу ебаться с БД, поэтому json).

main.py:

import asyncio

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

# импорт конфигов
from tgbot.config import load_config

# импорт регов мидлваря и хендлера /start
from tgbot.middlewares.data import DataMiddleware
from tgbot.handlers.command_start import register_command_start


# регистрация мидлваря
def register_all_middlewares(dp, data):
    dp.middleware.setup(DataMiddleware(data))


# регистрация хендлера
def register_all_handlers(dp):
    register_command_start(dp)

    
async def main():
    # чтение конфигов
    config = load_config('.env')
    
    # создаем экзепляры: диспетчер и бота
    storage = MemoryStorage()
    bot = Bot(token=config.bot.token, parse_mode=config.bot.parse_mode)
    dp = Dispatcher(bot, storage=storage)
    
    # регистрируем хендлер и мидлварь
    register_all_middlewares(dp, data=config.data)
    register_all_handlers(dp)
    bot.config = config
    
    # запуск пулинга
    try:
        print(f"Bot {config.bot.name} {config.bot.version} started!")
        await asyncio.gather(
            dp.start_polling()
        )
    finally:
        await bot.session.close()


if __name__ == '__main__':
    try:
        asyncio.run(main())
    except (KeyboardInterrupt, SystemExit):
        print(f'Bot stopped!')

Модуль для работы с JSON, он импортируется вместе с конфигом:

import json

class Data:

    def __init__(self):
        self.path = "tgbot/data/json/data.json" # путь к файлу
        self.storage = [] # хранилище
    
    # загружаем JSON в хранилище
    def load(self):
        with open(self.path) as file:
            self.storage = json.load(file)
            return self

    # сохранение хранилища в JSON
    def save(self):
        with open(self.path, 'w') as file:
            json.dump(self.storage, file)
            return True

    # добавление юзера в хранилище и сохранение
    async def add_user(self, user_id):
        self.storage.append(user_id)
        return self.save()

    # получаем True если юзер уже есть в хранилище
    async def get_user(self, user_id):
        return user_id in self.storage

    # получить список всех юзеров в хранинилище
    async def list_users(self):
        return self.storage

Регистрация мидлваря:

from aiogram.dispatcher.middlewares import LifetimeControllerMiddleware


class DataMiddleware(LifetimeControllerMiddleware):
    skip_patterns = ['error', 'update']
    
    # инициализация
    def __init__(self, data_obj):
        super().__init__()
        self.data = data_obj # экземпляр класса Data

    # класс Data - self.data, мы передали этот аргумент при инициализации как data_obj, теперь можем его юзать
    async def pre_process(self, obj, data, *args):
        data['data'] = self.data # передаем экземпляр Data в аргументах хендлеров

    # после исполнения хендлера удаляем объект
    async def post_process(self, obj, data, *args):
        del data['data']

По итогу мы передаем экземпляр Data при регистрации мидлваря, после чего можем с ним работать. Стоит отметить, что pre_process исполняется до исполнения хендлера, а вот post_process - после.


Хендлер для команды /start:

from aiogram import Dispatcher
from aiogram.types import Message
from aiogram.dispatcher import FSMContext

# импорт класса Data
from tgbot.data.data import Data


async def welcome(message: Message, data: Data, state: FSMContext):
    # проверяем наличие юзера в хранилище
    if not await data.get_user(message.from_user.id):
        # записываем юзера, если его нет
        await data.add_user(message.from_user.id)
        await message.answer(f"ID <b>{message.from_user.id}</b> записан.")
    # если такой юзер уже есть, то скипаем запись
    else:
        await message.answer(f"ID <b>{message.from_user.id}</b> уже был записан.")

# регистрируем хендлер
def register_command_start(dp: Dispatcher):
    dp.register_message_handler(welcome, commands=["start"], state='*')

Это все о чем я хотел рассказать, гайд был скорее для новичков, но все же.. на просторе рунета мало инфы. Так через мидлвари можно подключать БД, платежки и т.д. Подключайте тот элемент, который будет востребован на протяжении всего проекта, чтобы иметь к нему доступ из "любой" точки кода.

Глобальные переменные - полная дрисня, не юзай, побереги маму.

Спойлер: Скачать темплейт

VT: тык
Линк (ЯД): тык

 

[AGO]

В целом норм, но это зачем?) await asyncio.gather(
dp.start_polling()
)
Одна же всего функция

 

[SCOTE]

AGO, бля, это в других проектах юзал, у меня там несколько функций по дефолту, кроме пулинга. Забыл поменять.

 

[AGO]

SCOTE, за то что против глобалов закрепил твою статью. Делай ещё контент

 

[NiggaByte+REP]

сложна сложна нихуя не понятно

 

[AGO]

NiggaByte+REP, давай я сделаю и будет понятно сегодня утром

 

[NiggaByte+REP]

AGO, gooooooooooooooo