Начало

Привет, Хабр! Я учусь в Российско-Таджикском Славянском университете (на первом курсе), собственно у нас в университете действует так называемая кредитно-бальная система.

Для просмотра количества набранных баллов и так далее, у нас есть приложение которое было разработано университетом.

Оно доступно для Android.

Однако, я недавно перешёл на iOS-систему, собственно к моему удивлению приложения там не оказалось.

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

Да и вообще, Telegram работает везде, это моя любимая платформа для обмена сообщениями.

Поиск endpoint'ов...

Разумеется, университет сделал некий API для своего Android-фронтенда, для получения эндпоинтов не я, мой друг, декомпилировал APK файл и предоставил его мне, позже проанализировав "выхлоп" я нашел четыре необходимых мне эндпоинта

В частности эндпоинты для

• Авторизации

• Получения информации о профиле студента

• Получения семестров

• Получения данных об успеваемости по всем дисциплинам конкретного семестра.

Ну а дальше, дело за малым, надо просто написать клиент для этого API и так далее.

Инициализация проекта с помощью Poetry, написание обёртки под API

Создаём проект

poetry init 

Я сразу создал почти на самом верхнем уровне проекта пакетник rtsu где и будет лежать наша обёртка.

Давайте посмотрим на api.py

from aiohttp import ClientSession, ContentTypeError, client_exceptions
from cashews import cache
from typing import Optional, Union, Dict, TypeVar, Type, List, Self

from pydantic import BaseModel, parse_obj_as

from .exceptions import NotAuthorizedError, RtsuContentTypeError, ServerError, AuthError
from .schemas import AuthSchema, Profile, Subject, AcademicYear

RTSU_API_BASE_URL = "https://mobile.rtsu.tj/api/v1"
P = TypeVar("P", bound=BaseModel)


class RTSUApi:
    """
    This class provides for you functionality of RTSU public API
    """

    def __init__(self, token: Optional[str] = None):
        """
        Initializes `self`
        :param token: A rtsu-api token (optional)
        """

        self._api_token = token
        self._http_client = ClientSession()

    def set_token(self, token: str):
        """
        Setups token
        :param token: A token
        :return:
        """
        self._api_token = token

    async def _make_request(
            self,
            method: str,
            url_part: str,
            response_model: Type[Union[List[BaseModel], BaseModel]],
            json: Optional[Dict[str, Union[str, int]]] = None,
            params: Optional[Dict[str, str]] = None,
            auth_required: bool = False,
    ) -> Union[P, List[P]]:
        """
        Makes call to RTSU API
        :param url_part: Part of RTSU-API url, example - /auth
        :param json: A json for sending
        :param params: URI parameters for sending
        :return: Response object
        """

        if not json:
            json = {}

        if not params:
            params = {}

        headers = {}

        if auth_required:
            if not self._api_token:
                raise NotAuthorizedError("Not authorized, use `.auth` method.")

            headers['token'] = self._api_token

        try:
            response = await self._http_client.request(
                method,
                f"{RTSU_API_BASE_URL}/{url_part}",
                json=json,
                params=params,
                headers=headers,
                ssl=False,
            )
        except (client_exceptions.ClientConnectionError, client_exceptions.ClientConnectorError) as e:
            raise ServerError(f"Connection error, details: {e}")

        if response.status != 200:
            details = await response.text()
            raise ServerError(
                f"Server returned {response.status}, details: {details}"
            )

        try:
            deserialized_data = await response.json()
        except ContentTypeError as e:
            raise RtsuContentTypeError(
                e.message,
            )

        return parse_obj_as(response_model, deserialized_data)

    async def auth(self, login: str, password: str) -> AuthSchema:
        """
        Authenticates user
        :param login: A login of user
        :param password: A password of user
        :return: RTSU token on success
        """

        try:
            response: AuthSchema = await self._make_request(
                "POST",
                "auth",
                AuthSchema,
                params={
                    "login": login,
                    "password": password,
                }
            )
        except ServerError as e:
            raise AuthError(
                f"Auth error, check login and password, message from server: {e.message}"
            )

        self._api_token = response.token

        return response

    @cache.soft(ttl="24h", soft_ttl="1m")
    async def get_profile(self) -> Profile:
        """
        Returns profile of RTSU student
        :return: `Profile`-response
        """

        return await self._make_request(
            "GET",
            "student/profile",
            Profile,
            auth_required=True,
        )

    async def get_academic_years(self) -> List[AcademicYear]:
        """
        Returns `List` with `AcademicYear` objects
        :return:
        """

        return await self._make_request(
            "GET",
            "student/academic_years",
            List[AcademicYear],
            auth_required=True,
        )

    @cache.soft(ttl="24h", soft_ttl="1m")
    async def get_academic_year_subjects(self, year_id: int) -> List[Subject]:
        """
        Returns `List` with `Subjects` of some year
        :return:
        """

        return await self._make_request(
            "GET",
            f"student/grades/{year_id}",
            List[Subject],
            auth_required=True,
        )

    async def get_current_year_id(self) -> int:
        """
        Returns identifier of current year
        :return:
        """

        years = await self.get_academic_years()

        return years[0].id

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close_session()

    def __str__(self) -> str:
        """
        Stringifies `RTSUApi` objects
        :return:
        """

        return f"{self.__class__.__name__}<token={self._api_token}>"

    async def close_session(self):
        """Frees inner resources"""
        await self._http_client.close()

Что тут у нас? В _make_request у нас осуществляется запрос к серверу а также десериализация json в pydantic-схему (ну или модель?)

Прошу заметить, что я использую замечательную cashews для кеширования результатов, в частности soft-ttl который еще и сильно помогает когда сервера университета падают.

В остальных же методах я просто указываю эндпоинт и response-schema ну и дёргаю тот же _make_request

Также, тут есть методы для того чтобы закрыть текущую aiohttp-сессию, ну тут понятно, помимо этого реализованы магические методы __aenter__ и __aexit__ для того чтобы использовать клиент в withконструкциях.

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

Pydantic-схемы

Допустим, заглянем в profile.py где лежит Profile-schema

from pydantic import Field

from .base import Base


class FullName(Base):
    ru: str = Field(alias='RU')
    tj: str = Field(alias='TJ')


class Faculty(FullName):
    ...


class Speciality(FullName):
    ...


class Profile(Base):
    id: int = Field(alias='RecordBookNumber')
    full_name: FullName = Field(alias='FullName')
    faculty: Faculty = Field(alias="Faculty")
    course: int = Field(alias='Course')
    training_period: int = Field(alias='TrainingPeriod')
    level: str = Field(alias="TrainingLevel")
    entrance_year: str = Field(alias='YearUniversityEntrance')

Почему так? API возвращает мне информацию сразу на двух языках (русском и таджикском)

Собственно, это действительно свидетельствует о том, что API сделали очень криво и убого, но что поделать, тут я просто наследую каждый field от FullName чтобы не писать по два раза RU, TJ и так далее.

Также, прошу заметить то, как элегантно можно сделать field'ы pythonic при помощи pydantic-aliases

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

Собственно тесты к этому я тоже написал

import pytest
import pytest_asyncio

from rtsu_students_bot.rtsu import RTSUApi

from .config import settings

pytest_plugins = ('pytest_asyncio',)


@pytest_asyncio.fixture()
async def rtsu_client():
    """
    Initializes client
    :return: Prepared `RTSUApi` client
    """

    async with RTSUApi() as api:
        yield api


@pytest.mark.asyncio
async def test_rtsu_login(rtsu_client: RTSUApi):
    """
    Tests rtsu login
    :param rtsu_client: A RTSU API client
    :return:
    """

    resp = await rtsu_client.auth(settings.rtsu_api_login, settings.rtsu_api_password)

    assert resp.token is not None


@pytest.mark.asyncio
async def test_rtsu_profile_fetching(rtsu_client: RTSUApi):
    """
    Tests rtsu profile fetching
    :param rtsu_client:
    :return:
    """

    await rtsu_client.auth(settings.rtsu_api_login, settings.rtsu_api_password)

    profile = await rtsu_client.get_profile()

    assert profile is not None
    assert profile.full_name is not None


@pytest.mark.asyncio
async def test_rtsu_academic_years_fetching(rtsu_client: RTSUApi):
    """
    Tests rtsu academic years fetching
    :param rtsu_client:
    :return:
    """

    await rtsu_client.auth(settings.rtsu_api_login, settings.rtsu_api_password)

    years = await rtsu_client.get_academic_years()

    assert type(years) == list
    assert len(years) > 0


@pytest.mark.asyncio
async def test_rtsu_academic_year_subjects_fetching(rtsu_client: RTSUApi):
    """
    Tests rtsu academic year fetching
    :param rtsu_client:
    :return:
    """

    await rtsu_client.auth(settings.rtsu_api_login, settings.rtsu_api_password)

    ac_years = await rtsu_client.get_academic_years()
    year = ac_years[0].id
    years = await rtsu_client.get_academic_year_subjects(year)

    assert type(years) == list
    assert len(years) > 0

Тут тесты сделаны наверное грязно и тупо, я пока не читал про best-practises в тестировании API, буду рад предложениям в комментах по этому поводу.

Ах, да, для тестов тут используется отдельный конфиг, в конце статьи я покажу как его заполнить.

На руках у нас уже есть wrapper

База данных и SQLAlchemy

Установив алхимию, я принялся создавать пакет моделей

Но для начала, покажу вам файлик database.py в котором настроено подключение к базе.

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from rtsu_students_bot.config import settings

engine = create_async_engine(
    settings.db.url,
)

SessionLocal = sessionmaker(bind=engine, class_=AsyncSession)

Тут нет ничего такого, просто вместо обычной сессии я использую асинхронную (по очевидным причинам)

Вернёмся к пакетнику с моделями.

from sqlalchemy import Integer, Column, String, Boolean, BigInteger

from .base import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(length=255), nullable=True)
    token = Column(String(length=600), nullable=True)
    is_authorized = Column(Boolean, default=False)
    telegram_id = Column(BigInteger)

    def __str__(self):
        return f"{self.__class__.__name__}<id={self.id}, name={self.full_name}>"

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

Далее, пилим пакетник serviceкоторый будет помогать нам работать с базой данных

Содержимое user.py

from typing import Optional

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.models import User

from .exceptions import UserNotFound, UserAlreadyExists


async def get_user_by_tg_id(
        session: AsyncSession,
        telegram_id: int,
) -> Optional[User]:
    """
    Returns user by tg-id
    :param session: An `AsyncSession` object
    :param telegram_id: A telegram-ID
    :return: `User` or `None`
    """

    stmt = select(User).where(User.telegram_id == telegram_id)

    result = await session.execute(stmt)

    return result.scalars().first()


async def get_user_by_id(
        session: AsyncSession,
        user_id: int,
) -> Optional[User]:
    """
    Returns user by its id
    :param session: An `AsyncSession` object
    :param user_id: An ID
    :return: `User` or `None`
    """

    stmt = select(User).where(User.id == user_id)

    result = await session.execute(stmt)

    return result.scalars().first()


async def create_user(
        session: AsyncSession,
        telegram_id: int,
        full_name: Optional[str] = None,
        token: Optional[str] = None,
):
    """
    Creates `User` object
    :param session: An `AsyncSession` object
    :param telegram_id: A telegram-id
    :param full_name: Fullname of user
    :param token: A token of user
    :return: Created `User`
    """

    existed_user = await get_user_by_tg_id(session, telegram_id)

    if existed_user is not None:
        raise UserAlreadyExists(f"User with ID {telegram_id} already exists.")

    is_authorized = token is not None

    obj = User(
        telegram_id=telegram_id,
        full_name=full_name,
        token=token,
        is_authorized=is_authorized,
    )

    session.add(obj)
    await session.flush()
    await session.refresh(obj)

    return obj


async def update_user_token(
        session: AsyncSession,
        telegram_id: int,
        token: Optional[str] = None,
) -> User:
    """
    Authorizes `User`
    :param telegram_id:
    :param session:
    :param token:
    :return:
    """

    user = await get_user_by_tg_id(session, telegram_id)

    if not user:
        raise UserNotFound(f"User with telegram-id {telegram_id} not found.")

    is_authorized = token is not None

    stmt = update(User).where(
        int(user.id) == User.id
    ).values(
        is_authorized=is_authorized,
        token=token,
    )
    await session.execute(stmt)

    return await get_user_by_tg_id(session, user.telegram_id)


async def update_user(
        session: AsyncSession,
        user_id: int,
        telegram_id: Optional[int] = None,
        full_name: Optional[str] = None,
) -> User:
    """
    Updates telegram user
    :param session:
    :param user_id:
    :param telegram_id:
    :param full_name:
    :return:
    """

    user = await get_user_by_id(session, user_id)

    if user is None:
        raise UserNotFound(f"User with ID {user_id} not found.")

    stmt = update(User).where(User.id == user_id)

    if telegram_id is not None:
        stmt = stmt.values(
            telegram_id=telegram_id,
        )

    if full_name is not None:
        stmt = stmt.values(
            full_name=full_name
        )

    await session.execute(stmt)

    return await get_user_by_id(session, user_id)


async def delete_user(session: AsyncSession, user_id: int):
    """
    Deletes `User` object
    :param user_id:
    :param session: An `AsyncSession` object
    :return:
    """

    if await get_user_by_id(session, user_id) is None:
        raise ValueError("Invalid user-id passed.")

    stmt = delete(User).where(User.id == user_id)

    await session.execute(stmt)

Здесь я реализовал обычные функции которые будут помогать мне работать с базой, хотя наверное лучше бы я применил паттерн "Репозиторий"

Тестируем созданный CRUD

import pytest
import pytest_asyncio

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from rtsu_students_bot.service import user
from rtsu_students_bot.models import Base

from .config import settings

pytest_plugins = ('pytest_asyncio',)

engine = create_async_engine(
    settings.db_url,
)

SessionLocal = sessionmaker(autoflush=True, bind=engine, class_=AsyncSession)


@pytest_asyncio.fixture()
async def session():
    """
    Initializes client
    :return: Prepared `RTSUApi` client
    """

    async with SessionLocal() as e, e.begin():
        yield e


@pytest.mark.asyncio
async def test_tables_creating():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@pytest.mark.asyncio
async def test_user_creation(session: AsyncSession):
    """
    Tests user-creation
    :return:
    """

    user_data = {
        "full_name": "Vladimir Putin",
        "telegram_id": 1,
    }

    created_user = await user.create_user(session, **user_data)

    assert created_user.full_name == user_data.get("full_name")
    assert created_user.telegram_id == user_data.get("telegram_id")


@pytest.mark.asyncio
async def test_user_update(session: AsyncSession):
    """
    Tests user updating
    :param session:
    :return:
    """

    updating_data = {
        "full_name": "Volodymir Zelensky"
    }

    first_user = await user.get_user_by_tg_id(session, 1)

    updated_user = await user.update_user(session, first_user.id, **updating_data)

    assert first_user.id == updated_user.id
    assert first_user.telegram_id == updated_user.telegram_id
    assert updated_user.full_name == updating_data.get("full_name")


@pytest.mark.asyncio
async def test_user_token_updating(session: AsyncSession):
    """
    Tests user-token updating
    :param session:
    :return:
    """

    first_user = await user.get_user_by_tg_id(session, 1)

    assert not first_user.is_authorized

    first_user = await user.update_user_token(session, first_user.telegram_id, token="test token")

    assert first_user.is_authorized
    assert first_user.token == "test token"
    assert first_user.telegram_id == 1


@pytest.mark.asyncio
async def test_user_deleting(session: AsyncSession):
    """
    Tests user-token updating
    :param session:
    :return:
    """

    first_user = await user.get_user_by_tg_id(session, 1)

    assert first_user is not None

    await user.delete_user(session, first_user.id)

    first_user = await user.get_user_by_tg_id(session, 1)

    assert first_user is None

Тут также создается подключение к базе данных, я просто проверяю CRUD на работоспособность, собственно данные тесты помогли мне найти одну ошибку, поэтому я думаю писал я их не зря =)

Пилим самого бота

Для работы с Telegram Bot API я несомненно выбрал Aiogram ну и для этого сделал еще один пакетник.

Middlewares

Для того чтобы "протаскивать" необходимые ресурсы к хендлерам (API-клиент и AsyncSession), мне нужен специальный для этих целей мидлварь

import logging

from aiogram.dispatcher.middlewares import BaseMiddleware
from aiogram import types
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.rtsu import RTSUApi
from rtsu_students_bot.database import engine


class ResourcesMiddleware(BaseMiddleware):
    """
    Middleware for providing resources like db-connection and RTSU-client
    """

    def __init__(self):
        """
        Initializes self
        """

        self._logger = logging.getLogger("resources_middleware")

        super().__init__()

    @staticmethod
    async def _provide_api_client() -> RTSUApi:
        """
        Provides `RTSU` api client
        :return: Initialized client
        """

        client = RTSUApi()

        return client

    @staticmethod
    async def _provide_db_session() -> AsyncSession:
        """
        Provides `AsyncSession` object
        :return: Initialized session
        """

        session = AsyncSession(engine)

        return session

    async def _provide_resources(self) -> dict:
        """
        Initializes & provides needed resources, such as `RTSU-api-client` and `AsyncSession`
        :return:
        """
        self._logger.debug("Providing resources")
        api_client = await self._provide_api_client()
        db_session = await self._provide_db_session()

        resources = {
            "rtsu": api_client,
            "db_session": db_session,
        }

        return resources

    async def _cleanup(self, data: dict):
        """
        Closes connections & etc.
        :param data:
        :return:
        """

        self._logger.debug("Cleaning resources")

        if "db_session" in data:
            self._logger.debug("SQLAlchemy session detected, closing connection.")
            session: AsyncSession = data["db_session"]
            await session.commit()  # Commit changes
            await session.close()

        if "rtsu" in data:
            self._logger.debug("RTSU API Client detected, closing resource.")
            api_client: RTSUApi = data["rtsu"]
            await api_client.close_session()

    async def on_pre_process_message(self, update: types.Message, data: dict):
        """
        For pre-processing `types.Update`
        :param data: Data from other middlewares
        :param update: A telegram-update
        :return:
        """
        resources = await self._provide_resources()

        data.update(resources)

        return data

    async def on_pre_process_callback_query(self, query: types.CallbackQuery, data: dict):
        """
        Method for preprocessing callback-queries
        :param query: A callback-query
        :param data: A data from another middleware
        :return:
        """

        resources = await self._provide_resources()

        data.update(resources)

        return data

    async def on_post_process_callback_query(self, query: types.CallbackQuery, data_from_handler: list, data: dict):
        """
        Method for post-processing callback query
        :param data_from_handler: Data from handler
        :param query: A callback query
        :param data: A data from another middleware
        :return:
        """

        await self._cleanup(data)

    async def on_post_process_message(self, message: types.Message, data_from_handler: list, data: dict):
        """
        For post-processing message
        :param data_from_handler:
        :param message:
        :param data:
        :return:
        """
        await self._cleanup(data)

Тут все очень просто, при старте обработки сообщений/CallbackQuery я просто подгружаю ресурсы и передаю их в data

Также, после обработки всего этого дела, я вызываю _cleanup который при обнаружении сессий закроет их (ну и сделает коммит в случае с алхимией)

Получение пользователя

Для этого я сделал тоже отдельный мидлварь, который разумеется отрабатывает после первого

import logging

from aiogram.dispatcher.middlewares import BaseMiddleware
from aiogram import types
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.service import user
from rtsu_students_bot.rtsu import RTSUApi


class UserMiddleware(BaseMiddleware):
    """
    Middleware for providing a `User` object
    """

    def __init__(self):
        """
        Initializes self
        """

        self._logger = logging.getLogger("users_middleware")
        super().__init__()

    async def _provide_user(self, user_id: int, data: dict) -> dict:
        """
        Fetches and returns user
        """

        if 'db_session' not in data:
            raise RuntimeError("AsyncSession not found.")

        if 'rtsu' not in data:
            raise RuntimeError("RTSU API client not found.")

        db_session: AsyncSession = data.get("db_session")
        rtsu_client: RTSUApi = data.get("rtsu")

        self._logger.debug(f"Getting user with ID {user_id}")

        u = await user.get_user_by_tg_id(db_session, user_id)

        if u is None:
            self._logger.debug(f"User with ID {user_id} not found, creating...")
            u = await user.create_user(db_session, telegram_id=user_id)

        self._logger.debug(f"User provided, {u}")

        # If user is authorized, lets setup `RTSU` client
        if u.is_authorized:
            rtsu_client.set_token(u.token)
            self._logger.debug("User is authorized, API-client's token initialized.")

        data["user"] = u

        return data

    async def on_pre_process_message(self, message: types.message, data: dict):
        """
        Method for preprocessing messages (provides user)
        :param message: A message
        :param data: A data from another middleware
        :return: None
        """

        return await self._provide_user(message.from_user.id, data)

    async def on_pre_process_callback_query(self, query: types.CallbackQuery, data: dict):
        """
        Method for preprocessing callback-queries (provides user)
        :param data:
        :param query:
        :return:
        """

        return await self._provide_user(query.from_user.id, data)

Конкретно тут, мы просто создаем пользователя если на нашли его в бд, помимо того, если пользователь авторизован (токен есть в базе данных) мы также инициализируем токенAPI-клиента.

Теперь у нас есть почти всё что нужно для обработки сообщений.

Шаблонизатор Jinja2

Хорошо бы выделить "шаблоны" куда-то, собственно, я решил воспользоватся шаблонизатором "Нинздя"

Для этих целей я создал файлик template_engine.py

from typing import Optional, Any, Dict

from jinja2 import Environment, PackageLoader, select_autoescape

env = Environment(
    loader=PackageLoader('rtsu_students_bot', 'templates'),
    autoescape=select_autoescape(['html'])
)


def render_template(name: str, values: Optional[Dict[str, Any]] = None, **kwargs):
    """
    Renders template & returns text
    :param name: Name of template
    :param values: Values for template (optional)
    :param kwargs: Keyword-arguments for template (high-priority)
    """

    template = env.get_template(name)

    if values:
        rendered_template = template.render(values, **kwargs)
    else:
        rendered_template = template.render(**kwargs)

    return rendered_template

Тут я инициализировал шаблонизатор и сделал функцию render_template которая просто рендерит шаблон и возвращает готовый текст.

Шаблоны

Для них я сделал отдельную директорию в проекте, там ничего интересного я думаю вы не найдете.

Вот например шаблон для отправки профиля студента.

<b>???? Профиль</b>

<b>???? Полное имя: <em> {{ profile.full_name.ru }} </em></b>

<b>⚙️ ID Студента: <code>{{ profile.id }}</code></b>

<b>???? Факультет: <em>{{ profile.faculty.ru }} </em></b>

<b>ℹ️ Курс: <code>{{ profile.course }}</code></b>

<b>⏳ Период обучения: {{ profile.training_period }}{% if profile.training_period < 5 %} года {% else %} лет {% endif %}</b>

<b>???? Год поступления: {{ profile.entrance_year }}</b>

<b>???? Степень образования: {{ profile.level }} </b>

В общем, как-то так.

Бежим писать хендлеры

В core.py я положил общую функциональность которая может присутствовать в нескольких хендлерах, чтобы собственно соответсовать DRY

"""
`core.py` - Core-functionality of bot
"""

from typing import Union, List

from aiogram import types
from aiogram.dispatcher import FSMContext
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.service import user
from rtsu_students_bot.rtsu import RTSUApi, exceptions, schemas
from rtsu_students_bot.bot.keyboards import inline, reply
from rtsu_students_bot.bot.states import AuthState
from rtsu_students_bot.models import User
from rtsu_students_bot.template_engine import render_template


async def start_auth(
        update: Union[types.CallbackQuery, types.Message],
        user_in_db: User
):
    """
    Core-function, starts authentification process
    :param update: An `update` (message or query)
    :param user_in_db: An User in database
    :return:
    """

    markup = None
    text = render_template(
        "auth.html",
        user=user_in_db
    )

    if not user_in_db.is_authorized:
        await AuthState.first()
        markup = inline.cancellation_keyboard_factory()

    await update.bot.send_message(
        update.from_user.id,
        text=text,
        reply_markup=markup
    )


async def show_profile(
        message: types.Message,
        rtsu_client: RTSUApi,
        user_in_db: User,
):
    """
    Shows information about profile
    :param message:
    :param rtsu_client:
    :param user_in_db:
    :return:
    """

    profile = await rtsu_client.get_profile()

    text = render_template(
        "profile.html",
        profile=profile,
        user=user_in_db,
        telegram_user=message.from_user
    )

    await message.bot.send_message(
        message.from_user.id,
        text,
        reply_markup=inline.message_hiding_keyboard()
    )


async def show_statistics(
        message: types.Message,
        rtsu_client: RTSUApi,
):
    """
    Shows user's statistics
    :param message: A message
    :param rtsu_client: Initialized RTSU API client
    :return:
    """

    current_year_id = await rtsu_client.get_current_year_id()
    subjects: List[schemas.Subject] = await rtsu_client.get_academic_year_subjects(current_year_id)

    await message.bot.send_message(
        message.chat.id,
        text=render_template("statistics.html", subjects=subjects),
        reply_markup=inline.message_hiding_keyboard()
    )


async def authorize_user(
        update: Union[types.CallbackQuery, types.Message],
        user_in_db: User,
        login: str,
        password: str,
        db_session: AsyncSession,
        rtsu_client: RTSUApi,
        state: FSMContext
):
    """
    Authorizes user, on success auth, saves token to database
    :param state: A state (fsm-context)
    :param rtsu_client: An initialized RTSU api client
    :param db_session: `AsyncSession` object
    :param password: A password
    :param login: A login
    :param user_in_db: A user in database
    :param update: Update (message or query)
    """

    try:
        auth_schema = await rtsu_client.auth(login, password)
    except exceptions.AuthError:
        await update.bot.send_message(
            update.from_user.id,
            text=render_template("auth_error.html"),
            reply_markup=inline.cancellation_keyboard_factory()
        )
        await AuthState.first()
        return

    profile = await rtsu_client.get_profile()

    await user.update_user(
        db_session,
        user_in_db.id,
        full_name=profile.full_name.ru,
    )

    await user.update_user_token(
        db_session,
        update.from_user.id,
        auth_schema.token,
    )

    await update.bot.send_message(
        update.from_user.id,
        text=render_template("auth_success.html", full_name=profile.full_name.ru),
        reply_markup=reply.main_menu_factory()
    )
    await state.finish()


async def show_subjects(
        message: types.Message,
        rtsu_client: RTSUApi
):
    """
    Shows user's subjects
    :param message:
    :param rtsu_client:
    :return:
    """

    current_year = await rtsu_client.get_current_year_id()
    subjects = await rtsu_client.get_academic_year_subjects(current_year)

    await message.bot.send_message(
        message.chat.id,
        text="???? Дисциплины",
        reply_markup=inline.subjects_keyboard_factory(subjects)
    )


async def logout_user(
        message: types.Message,
        user_in_db: User,
        db_session: AsyncSession
):
    """
    Sets user's token to `NULL`
    :param db_session: `AsyncSession` object
    :param message: A message
    :param user_in_db: A user in db
    :return:
    """

    await user.update_user_token(
        db_session,
        user_in_db.telegram_id,
        token=None
    )

    await message.bot.send_message(
        message.from_user.id,
        text=render_template("logout.html", user=user_in_db),
        reply_markup=inline.auth_keyboard_factory()
    )


async def show_help(
        message: types.Message
):
    """
    Shows help-menu
    :param message: A message
    :return:
    """

    await message.bot.send_message(
        message.from_user.id,
        text=render_template("help.html"),
        reply_markup=inline.message_hiding_keyboard()
    )


async def show_about(
        message: types.Message
):
    """
    Shows about-menu
    :param message: A message
    :return:
    """

    await message.bot.send_message(
        message.from_user.id,
        text=render_template("about.html"),
        disable_web_page_preview=True,
        reply_markup=inline.message_hiding_keyboard()
    )

Тут вам и стейт-машина ну и клавиатурки собственно тоже есть.

Давайте заглянем непосредственно в хендлеры, в частности в commands.py

from aiogram import types, Dispatcher
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.rtsu import RTSUApi
from rtsu_students_bot.bot.filters import AuthorizationFilter
from rtsu_students_bot.models import User
from rtsu_students_bot.bot.keyboards import inline, reply
from rtsu_students_bot.template_engine import render_template

from . import core


async def start(message: types.Message, user: User):
    """
    Handles `/start` cmd
    :param user:
    :param message: A message
    """

    markup = reply.main_menu_factory()

    if not user.is_authorized:
        markup = inline.auth_keyboard_factory()

    await message.reply(
        text=render_template(
            "start.html",
            user=user,
            telegram_user=message.from_user
        ),
        reply_markup=markup,
    )


async def auth(message: types.Message, user: User):
    """
    Handles `/auth` cmd
    :param user: A User
    :param message: A message
    """

    await core.start_auth(message, user)


async def help_cmd(message: types.Message):
    """
    Handles `help` cmd
    :param message: A message
    """

    await core.show_help(message)


async def statistics(message: types.Message, rtsu: RTSUApi):
    """
    Handles `statistics` cmd
    :param message: A message
    :param rtsu: Initialized RTSU API client
    """

    await core.show_statistics(message, rtsu)


async def subjects(message: types.Message, rtsu: RTSUApi):
    """
    Handles `subjects` cmd
    :param message: A message
    :param rtsu: Initialized RTSU API client
    """

    await core.show_subjects(message, rtsu)


async def profile(message: types.Message, rtsu: RTSUApi, user: User):
    """
    Handles `profile` cmd
    :param message: A message
    :param user: User in db
    :param rtsu: Initialized RTSU API client
    """

    await core.show_profile(message, rtsu, user)


async def logout(message: types.Message, user: User, db_session: AsyncSession):
    """
    Handles `logout` cmd
    :param db_session: `AsyncSession` object
    :param user: A user in db
    :param message: A message
    """

    await core.logout_user(message, user, db_session)


async def about(message: types.Message):
    """
    Handles `about` cmd
    :param message: A message
    """

    await core.show_about(message)


def setup(dp: Dispatcher):
    """
    Setups commands-handlers
    :param dp:
    :return:
    """
    dp.register_message_handler(start, commands=["start"])
    dp.register_message_handler(help_cmd, commands=["help"])
    dp.register_message_handler(about, commands=["about"])
    dp.register_message_handler(logout, AuthorizationFilter(True), commands=["logout"])
    dp.register_message_handler(profile, AuthorizationFilter(True), commands=["profile"])
    dp.register_message_handler(subjects, AuthorizationFilter(True), commands=["subjects"])
    dp.register_message_handler(statistics, AuthorizationFilter(True), commands=["stat"])
    dp.register_message_handler(auth, AuthorizationFilter(False), commands=["auth"])
    dp.register_message_handler(auth, AuthorizationFilter(authorized=True), commands=["auth"])

Тут я просто регистрирую команды и дёргаю всё что мне нужно из core.py

Обратите внимание на AuthorizationFilter, он проверяет наличие авторизации.

Сразу возникает логичный вопрос, а почему я не использовал BoundFilter ?

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

Сообщать об этом разработчикам Aiogram - я не особо и имею желания, они в последнее время очень недружелюбны и очень странно относятся к русскоговорящим.

Меня например в чате aiogram тупо забанили :/

Собственно, привожу filters/auth.py и содержимое AuthorizationFilter

from typing import Union

from aiogram.dispatcher.filters import Filter
from aiogram.dispatcher.handler import ctx_data
from aiogram import types

from rtsu_students_bot.models import User
from rtsu_students_bot.template_engine import render_template


class AuthorizationFilter(Filter):
    """
    Filter for checking user's authorization
    """

    def __init__(self, authorized: bool):
        """
        Initializes self
        :param authorized:Is admin?
        """
        self.authorized = authorized

    async def check(self, message: Union[types.Message, types.CallbackQuery]):
        """
        Checks for user's authorization status
        :param message: A message
        """

        data = ctx_data.get()

        user: User = data.get("user")

        if self.authorized is None:
            return True

        if self.authorized and not user.is_authorized:
            await message.bot.send_message(
                message.from_user.id,
                text=render_template("not_authorized.html")
            )
            return False
        elif not self.authorized and user.is_authorized:
            await message.bot.send_message(
                message.from_user.id,
                text=render_template("already_authorized.html")
            )
            return False

        return True

Тут ничего сложного, я просто забираю данные из мидлваря, ну и если юзеру нужна авторизация но её - нет, я прерываю обработку апдейта вернув False, ну или наоборот если юзеру не нужна авторизация но она у него есть, я также прерываю обработку апдейта ну и не забываю выдавать ему сообщение, ничего такого.

Давайте взглянем на text.py и обработчики текста

"""
`text.py` - Text handlers
"""

from aiogram import types, Dispatcher
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters import Text
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.bot.filters import AuthorizationFilter
from rtsu_students_bot.rtsu import RTSUApi
from rtsu_students_bot.models import User
from rtsu_students_bot.bot.states import AuthState
from rtsu_students_bot.template_engine import render_template
from rtsu_students_bot.bot.keyboards import inline

from . import core


async def login_handler(
        message: types.Message,
        state: FSMContext
):
    """
    Handles `login` of user
    :param message: A message with login
    :param state: A current state (fsm-context)
    """

    async with state.proxy() as data:
        data["login"] = message.text

    await AuthState.next()
    await message.delete()

    await message.bot.send_message(
        message.from_user.id,
        render_template("enter_password.html"),
        reply_markup=inline.cancellation_keyboard_factory()
    )


async def password_handler(
        message: types.Message,
        state: FSMContext,
):
    """
    Handles password of user
    :param message: A message (with password)
    :param state: A state (fsm-context)
    """

    async with state.proxy() as data:
        password = data["password"] = message.text
        login = data["login"]

    await AuthState.next()
    await message.delete()

    await message.bot.send_message(
        message.from_user.id,
        render_template("credentials_confirmation.html", login=login, password=password),
        reply_markup=inline.confirmation_keyboard_factory()
    )


async def show_profile_handler(
        message: types.Message,
        rtsu: RTSUApi,
        user: User,
):
    """
    Handles 'Show profile' request
    :param message: A message
    :param rtsu: Initialized RTSU API client
    :param user: A user from db
    :return:
    """

    await core.show_profile(message, rtsu, user)


async def show_statistics_handler(
        message: types.Message,
        rtsu: RTSUApi,
):
    """
    Shows user's statistics
    :param message:
    :param rtsu:
    :return:
    """

    await core.show_statistics(message, rtsu)


async def show_subjects_handler(
        message: types.Message,
        rtsu: RTSUApi
):
    """
    Handles 'Show statistics' request
    :param rtsu: Initialized RTSU API client
    :param message: A message
    :return:
    """

    await core.show_subjects(message, rtsu)


async def logout_handler(
        message: types.Message,
        user: User,
        db_session: AsyncSession,
):
    """
    Handles 'logout-request'
    :param db_session: `AsyncSession` object
    :param message: A message
    :param user: A user in db
    """

    await core.logout_user(message, user, db_session)


async def auth_handler(message: types.Message, user: User):
    """
    Handles 'auth-request'
    :param message: A message
    :param user: A user in db
    :return:
    """

    await core.start_auth(message, user)


async def help_handler(message: types.Message):
    """
    Handles 'help-request'
    :param message: A message
    :return:
    """

    await core.show_help(message)


async def about_handler(message: types.Message):
    """
    Handles 'about-request'
    :param message: A message
    :return:
    """

    await core.show_about(message)


def setup(dp: Dispatcher):
    """
    Setups text-handlers
    :param dp: A `Dispatcher` instance
    """
    dp.register_message_handler(
        login_handler,
        state=AuthState.login,
        content_types=[types.ContentType.TEXT],
    )
    dp.register_message_handler(password_handler, state=AuthState.password, content_types=[types.ContentType.TEXT])
    dp.register_message_handler(
        show_profile_handler, Text(equals="???? Профиль"), AuthorizationFilter(authorized=True)
    )
    dp.register_message_handler(
        show_statistics_handler, Text(equals="???? Статистика"), AuthorizationFilter(authorized=True)
    )
    dp.register_message_handler(
        show_subjects_handler, Text(equals="???? Дисциплины"), AuthorizationFilter(authorized=True)
    )
    dp.register_message_handler(
        logout_handler, Text(equals="◀️ Выход из системы"), AuthorizationFilter(authorized=True)
    )
    dp.register_message_handler(
        auth_handler, Text(equals="???? Авторизация"), AuthorizationFilter(authorized=False)
    )
    dp.register_message_handler(
        help_handler, Text(equals="???? Инструкция"),
    )

    dp.register_message_handler(
        about_handler, Text(equals="ℹ️ О боте"),
    )

Тут мы обрабатываем кнопки главной менюшки и также дёргаем core.py для необходимого функционала, ну и помимо этого тут обрабатываются логин/пароль

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

Теперь взглянем на CallbackQuery-handlers

from aiogram import types, Dispatcher
from aiogram.dispatcher import FSMContext
from sqlalchemy.ext.asyncio import AsyncSession

from rtsu_students_bot.bot.filters import AuthorizationFilter
from rtsu_students_bot.rtsu import RTSUApi
from rtsu_students_bot.template_engine import render_template
from rtsu_students_bot.models import User
from rtsu_students_bot.bot.keyboards import callbacks, inline
from rtsu_students_bot.bot.states import AuthState

from .core import start_auth, authorize_user


async def auth_callback_processor(
        query: types.CallbackQuery,
        user: User
):
    """
    Handles `callbacks.AUTH_CALLBACK`
    :param query: A callback-query
    :param user: An User
    """

    await start_auth(query, user)


async def cancel_callback_processor(
        query: types.CallbackQuery,
        state: FSMContext
):
    """
    Handles `callbacks.CANCELLATION_CALLBACK`
    :param state: A current state (fsm-context)
    :param query:
    :return:
    """

    await query.message.delete()

    await query.bot.send_message(
        query.from_user.id,
        text=render_template("cancellation.html")
    )
    await state.finish()


async def credentials_confirmation_callback_processor(
        query: types.CallbackQuery,
        callback_data: dict,
        db_session: AsyncSession,
        user: User,
        rtsu: RTSUApi,
        state: FSMContext
):
    """
    Processes `callbacks.CONFIRMATION_CALLBACK`
    :param state: A current state (fsm-context)
    :param rtsu: An initialized rtsu-api client
    :param user: A user in database
    :param db_session: `AsyncSession` object
    :param query: A callback-query
    :param callback_data: Callback's data
    """

    # If user clicks `Yes` - `1` will be passed
    # If user clicks `No` - `0` will be passed
    # So, all data will be represented as strings in telegram-callbacks
    # For getting boolean some converting needed
    # Firstly, we convert string to int, after, we convert this int to boolean
    ok = bool(int(callback_data.get("ok")))

    await query.answer()
    await query.message.delete()

    async with state.proxy() as data:
        login = data.get("login")
        password = data.get("password")

    if ok:
        await authorize_user(query, user, login, password, db_session, rtsu, state)
    else:
        await query.bot.send_message(
            query.from_user.id,
            text=render_template("auth.html", user=user),
            reply_markup=inline.cancellation_keyboard_factory()
        )
        await AuthState.first()


async def show_subject_processor(
        query: types.CallbackQuery,
        rtsu: RTSUApi,
        callback_data: dict,
):
    """
    Handles `callbacks.SUBJECT_CALLBACK`
    :param callback_data: A callback-data
    :param query: A query
    :param rtsu: Initialized RTSU API client
    """

    await query.answer()

    needed_subject_id = int(callback_data.get("id"))

    year = await rtsu.get_current_year_id()

    subjects = await rtsu.get_academic_year_subjects(year)

    needed_subject = list(filter(lambda x: x.id == needed_subject_id, subjects))

    if not needed_subject:
        await query.bot.send_message(
            query.from_user.id,
            "Дисциплина не найдена."
        )
        return

    await query.bot.send_message(
        query.from_user.id,
        text=render_template(
            "subject.html",
            subject=needed_subject[0]
        )
    )


async def delete_message_callback_processor(query: types.CallbackQuery):
    """
    Processes deletion-callback
    :param query: A callback-query
    :return:
    """
    await query.answer()
    await query.message.delete()


def setup(dp: Dispatcher):
    """
    Registers callback-query handlers
    :param dp: A `Dispatcher` instance
    """
    dp.register_callback_query_handler(
        auth_callback_processor, callbacks.AUTH_CALLBACK.filter(), AuthorizationFilter(False)
    )
    dp.register_callback_query_handler(cancel_callback_processor, callbacks.CANCELLATION_CALLBACK.filter())
    dp.register_callback_query_handler(
        credentials_confirmation_callback_processor, callbacks.CONFIRMATION_CALLBACK.filter(), state=AuthState.confirm
    )
    dp.register_callback_query_handler(
        show_subject_processor, callbacks.SUBJECT_CALLBACK.filter(), AuthorizationFilter(True)
    )
    dp.register_callback_query_handler(
        delete_message_callback_processor, callbacks.DELETE_MSG_CALLBACK.filter()
    )

Тут также дёргается некоторый функционал из core.py но помимо этого, есть и обработчики которые по сути самостоятельные и выдают какую-то информацию.

Тут вам и сабмиты логинов/паролей и выдача отчёта по конкретной дисциплине ну и так далее всё, что связано с inline-клавиатурой.

В целом, есть еще и обработчики ошибок, вот они

import logging

from aiogram import Dispatcher, types
from aiogram.utils.exceptions import InvalidQueryID, MessageNotModified

from rtsu_students_bot.rtsu.exceptions import ServerError
from rtsu_students_bot.template_engine import render_template


async def invalid_query_error_handler(update, error):
    """
    Handles `InvalidQueryError`
    :param update:
    :param error:
    :return:
    """
    logging.info(f"OK, Invalid query ID, {error}, {update}")
    return True


async def message_not_modified_error_handler(update, error):
    """
    Handles `MessageNotModifiedError`
    :param update:
    :param error:
    :return:
    """
    logging.info(f"OK, Message not modified, {error}, {update}")
    return True


async def server_error_handler(update: types.Update, error: ServerError):
    """
    Handles `ServerError`
    :param update:
    :param error:
    :return:
    """

    logging.exception("Server error", exc_info=error)

    if update.message:
        chat_id = update.message.from_user.id
    else:
        chat_id = update.callback_query.from_user.id

    await update.bot.send_message(
        chat_id=chat_id,
        text=render_template("server_error.html")
    )

    return True


async def any_exception_handler(update: types.Update, error: Exception):
    """
    Handles `Exception`
    :param update:
    :param error:
    :return:
    """
    logging.error(f"{error.__class__.__name__} has been thrown")
    logging.exception("Exception", exc_info=error)
    return True


def setup(dp: Dispatcher):
    """
    Registers error handlers
    :param dp: A `Dispatcher` instance
    """
    dp.register_errors_handler(
        server_error_handler,
        exception=ServerError
    )
    dp.register_errors_handler(
        invalid_query_error_handler,
        exception=InvalidQueryID
    )
    dp.register_errors_handler(
        message_not_modified_error_handler,
        exception=MessageNotModified
    )
    dp.register_errors_handler(
        any_exception_handler,
        exception=Exception
    )

Тут мы просто обрабатываем и пропускаем некоторые ошибки, а об ServerError мы сообщаем пользователю.

Думаю тоже логично.

Теперь взглянем на главный файл в пакете bot, а именно app.py

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

from rtsu_students_bot.config import settings

from . import handlers, middlewares


def get_app() -> Dispatcher:
    """
    Initializes & returns `Dispatcher`
    """

    # Create bot & dispatcher

    memory_storage = MemoryStorage()

    bot = Bot(settings.bot.token, parse_mode="html")
    dp = Dispatcher(bot, storage=memory_storage)

    # Setup handlers

    handlers.setup(dp)

    # Setup middlewares

    middlewares.setup(dp)

    return dp

Тут мы просто создаем объекты диспетчера и бота, регистрируем обработчики, мидлвари ну и задаем storage для нашего FSM

Да, у меня пока что всё хранится в MemoryStorage, возможно позже сделаю RedisStorage, увидим.

Ну и на последок, cli.py & config.py

from aiogram import executor
from typer import Typer, Option

from .config import settings
from .bot import get_app, handlers

typer_app = Typer()


@typer_app.command()
def start(
        skip_updates: bool = Option(default=False, help="Skip telegram updates on start?"),
        use_webhook: bool = Option(default=False, help="Use webhook for receiving updates?")
):
    """
    Starts bot
    :param skip_updates: Skip telegram updates on start?
    :param use_webhook: Use webhook mode for receiving updates?
    """

    # Build bot
    dp = get_app()

    # Build startup-handler

    startup_handler = handlers.startup.startup_handler_factory()

    if use_webhook:
        # Check for `webhook` settings are not `None`

        if settings.webhooks is None:
            print("Please, fill webhook's settings.")
            exit(-1)

        startup_handler = handlers.startup.startup_handler_factory(f"{settings.webhooks.host}{settings.webhooks.path}")

        executor.start_webhook(
            dispatcher=dp,
            on_startup=startup_handler,
            skip_updates=skip_updates,
            host=settings.webhooks.webapp_host,
            port=settings.webhooks.webapp_port,
            webhook_path=settings.webhooks.path,
            check_ip=True
        )
    else:
        executor.start_polling(
            dp,
            skip_updates=skip_updates,
            on_startup=startup_handler
        )

Тут у меня просто используется Typer для обработки аргументов командной строки, в целом тут я просто подгружаю конфиг и запускаю бота, startup-хендлеры я думаю вы посмотрите уже сами, если интересно, там просто настройка логов, бд, кеша и всего остального, вряд-ли вам это покажется интересным.

Ну и собственно config.py

from typing import Optional

from pydantic import BaseSettings

from .constants import DEFAULT_ENCODING, SETTINGS_FILE


class DatabaseSettings(BaseSettings):
    """Settings of database"""
    url: str


class BotSettings(BaseSettings):
    """Settings of telegram-bot"""
    token: str


class Logging(BaseSettings):
    format: str
    debug: bool


class Webhooks(BaseSettings):
    host: str
    path: str
    webapp_host: str
    webapp_port: int


class Settings(BaseSettings):
    """Class for settings"""
    bot: BotSettings
    logging: Logging
    db: DatabaseSettings
    webhooks: Optional[Webhooks] = None


settings = Settings.parse_file(
    path=SETTINGS_FILE,
    encoding=DEFAULT_ENCODING
)

Тут тоже ничего такого нет, использую обычные настройки из Pydantic и гружу их в settings

Модули в питоне кешируются, поэтому такой вид конфигурации считаю в принципе справедливым =)

Ну и собственно, скриншоты получившегося бота

Собственно, на этом думаю всё, все ссылки прикрепляю.

• Исходный код проекта.

• Ссылка на бота

• Мой телеграм

Использованные библиотеки/фреймворки

• Aiogram

• Pydantic

• Aiohttp

• Cashews

• SQLAlchemy

• Poetry

• Typer

• Jinja2

Спасибо что прочитали до конца, ожидаю объективной критики и предложений по улучшению качества кода для своего же развития.