В данной, как и в любой другой, публикации я рассмотрю разработку телеграмм ботов на новой версии aiogram 3.x. А точнее разработку бота, асинхронные запросы к postgresql,асинхронные запросы по http, да и в принципе полезное использование asyncio, если вдруг у вас произойдёт асинхронность головного мозга, как у меня.

Вступление

В мире современных технологий и динамичных информационных потоков студенты и преподаватели ищут удобные и эффективные способы взаимодействия. В этом контексте создание телеграмм‑бота для университета становится настоящим спасением, объединяющим удобство и функциональность в одном лице. Сегодня я расскажу вам, почему именно эти технологии — aiogram 3.x, aiohttp, asyncio и asyncpg (PostgreSQL) — стали основой для разработки нашего инновационного помощника.

Во‑первых, aiogram 3.x — это мощный и гибкий фреймворк для создания телеграмм‑ботов на Python. С его помощью мы можем быстро и легко реализовать сложные сценарии взаимодействия, предоставляя пользователям интуитивно понятный интерфейс. Aiogram позволяет нам сосредоточиться на логике бота, не утопая в технических деталях.

Далее у нас aiohttp — это высокопроизводительный HTTP‑клиент и сервер для Python. Он идеально подходит для обработки запросов и взаимодействия с внешними API. С aiohttp наш бот может получать актуальные данные из университетских систем, отправлять уведомления и обеспечивать бесперебойное общение в реальном времени.

Но что связывает все это вместе? Ответ — asyncio. Этот модуль позволяет писать асинхронный код, обеспечивая высокую производительность и отзывчивость. С asyncio наш бот может одновременно обрабатывать множество запросов, не теряя при этом скорости и надежности. Это значит, что пользователи будут получать мгновенные ответы на свои вопросы, даже в периоды высокой нагрузки.

И наконец, asyncpg — это асинхронный драйвер для PostgreSQL. Эта база данных славится своей мощностью и надежностью. С помощью asyncpg наш бот может быстро и безопасно работать с большими объемами данных, обеспечивая студентов и преподавателей актуальной информацией в любое время.

Таким образом, использование связки aiogram 3.x, aiohttp, asyncio и asyncpg (PostgreSQL) делает нашего телеграмм‑бота не просто функциональным, но и невероятно эффективным. Он становится вашим надежным помощником в учебном процессе, всегда готовым прийти на помощь, ответить на вопросы и в принципе упростить жизнь в университете.

Вот оно будущие🤖

Куда поставить

Если что то не стоит на своем месте, то это нужно исправить. В случае нашего бота я буду использовать сервис AMVERA, поскольку этот сервис как удобен в эксплуатации вообще, так и очень удобен в тестировании приложения, т.к. у них есть почасовая оплата серверов, ну и конечно же 100р в подарок не могут не радовать.

Все инструкции как поставить бота на хостинг и развернуть базу данных есть на их сайте или тут, тут.

И после того как с бесплатной рекламой покончено можно перейти дальше.

Немного о боте

В этом проекте мы в основном сосредоточились на недостатках которые присутствуют в получении расписания студентами, что зачастую является не очень удобным, хотя в нашем вузе имеется целых 3 способа получения расписания : календарь(в том числе интеграция в google календарь), studenfy,официальный сайт. Но нам показалась мало и мы захотели сделать бота с не очень больший функционалом, но с получением уведомление с расписанием каждую неделю, каждый день и каждую пару.

Парсинг

Поскольку обычно нужно было бы парсить расписание с сайта, но в нашем случае есть замечательный api с которого можно удобно парсить мы будем использовать aiohttp.

Для получения расписания можно сделать запрос на этот адрес https://parser.ystuty.ru/api/ystu/schedule/group/{group} который выдаст расписание на ВЕСЬ семестр.

запрос я делал так:

async def get_scheld(group):
    """
    асинхронно отправляет запрос к апи с группой,возвращает словарь
    :param group: группа для которой нужно получить расписание
    :return: словарь с расписанием
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://parser.ystuty.ru/api/ystu/schedule/group/{group}") as response:
            return await response.json()

Этот код использует библиотеку aiohttp для выполнения асинхронного HTTP-запроса и получения данных в формате JSON. Давайте разберем его по шагам:

1. Создание асинхронной сессии:

   async with aiohttp.ClientSession() as session:

   Этот блок создаёт асинхронную сессию HTTP-запросов, используя контекстный менеджер async with. Это обеспечивает автоматическое закрытие сессии после завершения блока.

2. Отправка асинхронного GET-запроса:

   async with session.get(f"https://parser.ystuty.ru/api/ystu/schedule/group/{group}") as response:

   Внутри контекстного менеджера сессии, этот блок выполняет GET-запрос по указанному URL. Переменная group подставляется в URL-адрес. Результат запроса сохраняется в переменную response.

3. Получение и возврат данных в формате JSON:

   return await response.json()

   Этот блок асинхронно извлекает JSON-данные из ответа и возвращает их.

Поскольку расписание приходит в json, я перевожу его в словарь для удобства работы.

Для расписания на сегодняшний день, расшифрованное сообщение будет выглядеть примерно так:

async def scheld_today(group):
    """
    асинхронно из полученного словаря получает расписание на сегодня по группе
    :param group:группа к каторой получает расписание
    :return: словарь расписание на сегодня,если распиания нет то None
    """
    try:
        t = await get_scheld(group)
        for j in range(len(t['items'])):
            for i in range(len(t['items'][j]['days'])):
                dt = (t['items'][j]['days'][i]['info']['date']).split("T")[0]
                res=0
                if dt == str(date.today()):
                    res =t['items'][j]['days'][i]
                    break
            if res!=0:
                break
        return res
    except :
        return None

Этот код возвращает расписание в виде словаря, если расписание есть, и самый желаемый результат: если расписания нет, то возвращает None.

В принципе этого уже хватит для многого, но это только начало, поскольку этот текст нужно еще отправить.

Для формирования я использовал примерно такой код:

sch = await scheld_today(group)
if not((sch ==0) or (sch == None)):
  lessons = "".join([f"{i['originalTimeTitle']} | {i['lessonName']}\n{i['auditoryName']} | {'Неизвестно' if i['teacherName'] is None else i['teacherName']}\n" for i in sch['lessons']])
  sch_ = f"{sch['info']['name']}\nПары на день:\n"+lessons
else:
  sch_ = "на расслабоне🎆"

Конечно сообщение по дизайну хуже чем что-либо, но оно информативно и удобно, что мы и пытаемся достичь.

Работа с бд

Использование asyncpg и PostgreSQL в асинхронном Telegram-боте обладает несколькими важными преимуществами:

1. Высокая производительность: asyncpg - это высокопроизводительный асинхронный клиент для PostgreSQL, который обеспечивает быструю обработку запросов к базе данных без блокировки основного потока выполнения. Asyncpg оказывается (по заявлению создателей) в среднем в 3 раза быстрее, чем psycopg2 (или aiopg).

2. Совместимость с асинхронными фреймворками: asyncpg хорошо интегрируется с популярными асинхронными фреймворками Python, такими как aiohttp и aiogram, что упрощает создание полностью асинхронного стека приложений. А также использовать все возможности асинхронного подхода.

В основании своем asyncpg также отправляет запросы к базе данных с помощью SQL.

Код для отправки запроса к бд у меня такой:

async def async_db_request(query, params):
  """
  Асинхронная функция для выполнения запроса к базе данных postgresql
  Args:
      query: Текст запроса
      params: Словарь с параметрами запроса
  Returns:
      Результат запроса
  """
  conn = await asyncpg.connect(
      database=database,
      user=user,
      password=pas,
      host=host,
      port=port)
  try:
    if params is None:
      result = await conn.fetch(query)
    else:
      result = await conn.fetch(query, params)
  except Exception as e:
    raise RuntimeError(f"Ошибка при запросе к базе данных: {e}")
  finally:
    await conn.close()
  return result

В принципе данная функцию отправляет любой запрос к бд, и любые необходимые запросы к бд можно отправлять просто вызываю данную функцию и прописав запрос с помощью SQL

Выглядит это примерно так:

async def new_review(rev,them,who):
    """
    добавляет новый отзыв в базу данных
    :param rev: сам отзыв
    :param them: тема отзыва
    :param who: кто отправил отзыв
    """
    await async_db_request(f"INSERT INTO reviews (review,theme,who_send) VALUES ('{rev}','{them}','{who}');",params=None)

Ну и в принципе статей на тему asyncpg очень много, поэтому можно идти дальше.

Про aiogram 3

В этой главе немного расскажу про особенности, которые я заметил при работе с новой версией асинхронграмма. По большей мере принципы остались примерно такие же, однако были введены изменения которые улучшили или упростили, по моему мнению, работу с библиотекой.

Router - позволяет гораздо легче разбить бота на множество под ботов в отдельных файлах, что улучшает структурирование проекта. В общем случаем новый роутер обозначается и используется примерно так :

from aiogram import Router
my_router = Router()
@my_router.message()

но также роутер нужно добавить в диспатчер:

# Объект бота
bot = Bot(token=token)
# Диспетчер
dp = Dispatcher()
#подключение отдельных диспатчеров
dp.include_router(my_router)

Главное не забывать, что роутеры используют иерархию, точнее каждый родительский бот перехватывает вызовы первее, а нижнее не реагируют. На картинке с подробного гайда это выглядит так:

F - респект, а в нашем случае фильтры.

Стандартный F.text, но немного с магией. Как всегда обрабатывает просто текст, примеры:

F.text == 'hello'  # lambda message: message.text == 'hello'
F.text != 'spam'  # lambda message: message.text != 'spam'

А также:

F.text.startswith('foo')  # lambda message: message.text.startswith('foo')
F.text.endswith('bar')  # lambda message: message.text.startswith('bar')
F.text.contains('foo')  # lambda message: 'foo' in message.text

Да даже фото,голоса и даже несколько фильтров:

F.photo                                    # Фильтр для фото
F.voice                                    # Фильтр для голосовых сообщений
F.content_type.in_({ContentType.PHOTO,
                    ContentType.VOICE,
                    ContentType.VIDEO})    # Фильтр на несколько типов контента

Что является очень удобным и практичным для большинства случаев, хотя и можно просто использовать лямбда выражения, а также создавать свои собственные классы для фильтров, как например такой код который проверяет тип чата:

from typing import Union

from aiogram.filters import BaseFilter
from aiogram.types import Message


class ChatTypeFilter(BaseFilter):  # [1]
    def __init__(self, chat_type: Union[str, list]): # [2]
        self.chat_type = chat_type

    async def __call__(self, message: Message) -> bool:  # [3]
        if isinstance(self.chat_type, str):
            return message.chat.type == self.chat_type
        else:
            return message.chat.type in self.chat_type

Еще в новой версии добавили небольшие изменения кнопок и всего остального, но по большему счету эти изменения не сильно затрагивались в проекте, поэтому разбирать их я не вижу смысла (

asyncio - бате всех библиотек

Использование asyncio в телеграм-боте, который взаимодействует с базой данных и парсит информацию из интернета, предоставляет множество преимуществ. Во-первых, asyncio позволяет эффективно управлять асинхронными операциями, что особенно важно для телеграм-ботов, которым необходимо обрабатывать большое количество запросов в реальном времени. Асинхронный подход позволяет не блокировать основной поток выполнения во время выполнения операций ввода-вывода, таких как запросы к базе данных или парсинг веб-страниц. Это, в свою очередь, повышает производительность и отзывчивость бота, делая его более эффективным и быстрым в обработке запросов пользователей.

Во-вторых, использование asyncio упрощает управление несколькими задачами одновременно. Например, телеграм-бот может одновременно обрабатывать входящие сообщения, выполнять запросы к базе данных и парсить данные из интернета без необходимости ожидания завершения одной задачи перед началом другой. Это позволяет оптимально использовать ресурсы системы и минимизировать задержки, улучшая пользовательский опыт.

Кроме того, asyncio предоставляет гибкие инструменты для управления ошибками и исключениями в асинхронных операциях, что упрощает разработку и отладку сложных асинхронных приложений. Возможность использовать async/await синтаксис делает код более читаемым и поддерживаемым по сравнению с традиционными подходами к многозадачности, такими как многопоточность.

В нашем боте мы активно используем asyncio для разделения задач отправки уведомлений и основного обработчика, выглядит это примерно так:

начинается это в main.py, когда создаются 2 задачи, обработчика и уведомлений:

async def main():
    polling_task = asyncio.Task(dp.start_polling(bot))
    my_async_function_task = asyncio.Task(notify(bot,tst=True))

    await asyncio.gather(polling_task, my_async_function_task)

if __name__ == "__main__":
    asyncio.run(main())

Это создает 2 асинхронных задачи которые работают независимо друг от друга, это позволяет правильно распределить нагрузку и беспрепятственно работать основному диспетчеру, что нам необходимо.

В раздели обновлений запускаются в 3 часа ночи еще 3 асинхронных функции, которые отвечают за отправку уведомлений в свое время, это необходимо т.к. я усыпляю их до необходимого момента, в который они включаются и отправляют уведомления. В виде кода это выглядит так:

async def notify(bot, tst=False):
        time = datetime.now()
        time = time.second + time.minute * 60 + time.hour * 3600
        next_time = rest_time * 3600 + 60 - time
        logging.info(f"След. время обновления расписания через {next_time / 3600}")

        await asyncio.sleep(next_time)
        time_w = datetime.today().weekday()

        await evd_sch()
        await evl_sch()
        logging.info(f"---Данные обновлены---")

        if time_w == 0:
            await evw_sch()
            await asyncio.gather(evd(bot, tst),evl(bot, tst),evw(bot, tst))
        else:
            await asyncio.gather(evd(bot, tst), evl(bot, tst))

        logging.info("Уведомления успешно отправлены")
        t=300
        await asyncio.sleep(t)
        await notify(bot, tst)

Здесь также можно увидеть await asyncio.sleep()_, который усыпляет только ту асинхронную функцию, в которой он был запущен, поэтому наш основной диспетчер продолжает спокойно работать. А в самой функции notif запускаются 3 асинхронные функции (с помощью создания задач await asyncio.gather()),которые также асинхронно засыпают, и после их запуска главная функция перезапускается и ждет своего времени.

На этом микро разбор применения asyncio в телеграмм боте закончен(

Вывод

На данном мой небольшой разбор применения различных асинхронных библиотек для создания телеграмм бота закончен, разборы конкретно по каждой библиотеке можно легко найти в интернете, поскольку мне лень их писать. А так надеюсь эта статья кому-нибудь будет полезна.

Powodzenia wszystkim i na razie!