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

Для полного погружения желательно, чтобы вы уже имели базовые знания Python, были знакомы с фреймворком Aiogram 3 (на моем Хабре уже есть около 15 публикаций, в которых я подробно разбираю создание телеграм‑ботов с нуля на этом фреймворке), а также имели общее представление о базах данных, в частности SQLite, и их интеграции с Python.

Что мы будем делать сегодня?

Сегодня мы создадим телеграм-бота для хранения заметок и файлов. Мы будем использовать фреймворк Aiogram 3 для разработки, а базу данных SQLite с асинхронным движком aiosqlite для хранения данных. Наш бот будет иметь следующий функционал:

• Добавление заметок с любым содержимым: текст, фото, видео, аудио, голосовые сообщения и т. д.

• Удаление заметок

• Редактирование текстового содержимого заметок

• Удобный поиск заметок по текстовому содержимому, дате добавления и типу контента

Особенности нашего бота

1. Хранение медиа на серверах Telegram. Всё мультимедийное содержимое (фото, видео, документы и т. д.) мы будем хранить прямо на серверах Telegram, что позволит существенно экономить пространство и ресурсы на вашем сервере. На своей стороне мы будем хранить только айди этих файлов.

2. Использование SQLAlchemy с aiosqlite. Для взаимодействия с базой данных мы будем использовать SQLAlchemy — это гибкий и мощный ORM, который, несмотря на пугающие стереотипы, весьма прост в использовании, особенно в контексте ботов. Мы задействуем асинхронный движок aiosqlite для работы с базой данных SQLite, что позволит сделать бота более отзывчивым.

3. Деплой в облако. Чтобы бот работал не только локально на вашем компьютере, но и в облаке, мы развернём его на платформе Amvera Cloud. Этот сервис предлагает удобный способ развертывания проектов. Всё, что нужно для деплоя — создать файл конфигурации, который я вам предоставлю в разделе про деплой. Затем вы сможете загрузить этот файл вместе с файлами бота на сервис. Это можно сделать как через Git, так и напрямую через внутреннюю консоль Amvera Cloud. После загрузки файлов на сервис проект автоматически соберется и запустится.

Подготовка

Перед началом работы убедитесь, что у вас есть базовые навыки программирования на Python, и вы понимаете, как работают телеграм-боты. Для продолжения разработки вам понадобится токен вашего телеграм-бота, который можно получить через BotFather, следуя следующей инструкции:

1. Откройте чат с BotFather в Telegram.

2. Введите команду /newbot.

3. Следуйте указаниям для создания нового бота.

4. Сохраните токен, который вам выдаст BotFather — он понадобится для интеграции с вашим кодом.

Подготовим файл requirements.txt и заполним его следующими библиотеками:

aiosqlite==0.20.0
aiogram==3.12.0
python-decouple==3.8
sqlalchemy==2.0.35

Сегодня нам понадобятся именно эти библиотеки:

• Aiosqlite — асинхронный движок для работы с базами данных, который мы будем использовать вместе с SQLAlchemy.

• Aiogram — библиотека для создания ботов на платформе Telegram.

• Python‑decouple — библиотека для работы с переменными окружения, что позволяет удобно управлять конфигурацией проекта.

• SQLAlchemy — мощный ORM (Object‑Relational Mapping) для работы с базами данных.

Для установки этих библиотек выполните следующую команду:

pip install -r requirements.txt

На данном этапе у вас уже должен быть установлен Python, настроен проект в вашей IDE с виртуальным окружением и необходимыми библиотеками, а также готов токен для работы с Telegram API. Если всё это у вас есть, давайте приступим к написанию кода!

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

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

Для начала создадим пакет (папку с файлом __init__.py), в которую будем помещать все файлы, которые будут иметь отношение к базе данных бота и взаимдействию с ней.

Я назову пакет data_base, но имя может быть любым.

Давайте создадим внутри файл database.py. Данный файл можно прировнять к главному конфигурационному классу базы данных.

Напишем код, а после разберемся что он делает.

from sqlalchemy import func
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession


engine = create_async_engine(url='sqlite+aiosqlite:///db.sqlite3')
async_session = async_sessionmaker(engine, class_=AsyncSession)


class Base(AsyncAttrs, DeclarativeBase):
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

Этот код настраивает асинхронное взаимодействие с базой данных SQLite с использованием SQLAlchemy и определяет базовый класс для ORM-моделей. Вот краткое описание каждой части:

1. engine = create_async_engine(...): Создает асинхронный движок для работы с базой данных SQLite через протокол aiosqlite. Этот движок управляет подключениями к базе данных.

2. async_session = async_sessionmaker(...): Определяет фабрику для создания асинхронных сессий работы с базой данных. Эти сессии используются для выполнения запросов и операций с базой.

3. class Base(AsyncAttrs, DeclarativeBase): Это базовый класс для всех ORM‑моделей. Он наследует:

   • AsyncAttrs: добавляет поддержку асинхронных операций для моделей.

   • DeclarativeBase: базовый класс, который определяет декларативный стиль работы с SQLAlchemy (когда модели описываются как классы Python).

4. created_at и updated_at:

   • created_at: колонка для хранения времени создания записи. Значение по умолчанию устанавливается с помощью функции func.now(), которая генерирует текущую дату и время.

   • updated_at: колонка для времени последнего обновления записи. Также используется func.now(), но с параметром onupdate=func.now(), который автоматически обновляет время при каждом изменении записи.

Для PostgreSQL данный файл имел бы похожую структуру, за исключением другой ссылки для подключения и асинхронного движка asyncpg.

Создаем модели

Теперь подготовим модели таблиц в SQLAlchemy. Модель в этом контексте — это класс, представляющий таблицу в базе данных. С одной стороны, мы описываем сам класс, а с другой — каждая колонка таблицы определяется как отдельный атрибут, с которым можно работать как с объектом. Таким образом, модель выступает в роли мостика между объектами Python и данными, хранящимися в базе.

Модели мы опишем в файле models.py

from sqlalchemy import BigInteger, Integer, Text, ForeignKey, String
from sqlalchemy.orm import relationship, Mapped, mapped_column
from .database import Base


# Модель для таблицы пользователей
class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    username: Mapped[str] = mapped_column(String, nullable=True)
    full_name: Mapped[str] = mapped_column(String, nullable=True)

    # Связи с заметками и напоминаниями
    notes: Mapped[list["Note"]] = relationship("Note", back_populates="user", cascade="all, delete-orphan")


# Модель для таблицы заметок
class Note(Base):
    __tablename__ = 'notes'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), nullable=False)
    content_type: Mapped[str] = mapped_column(String, nullable=True)
    content_text: Mapped[str] = mapped_column(Text, nullable=True)
    file_id: Mapped[str] = mapped_column(String, nullable=True)
    user: Mapped["User"] = relationship("User", back_populates="notes")

Эти модели описывают две связанные таблицы в базе данных: пользователей (User) и заметок (Note). Вот краткий разбор:

Модель User

• __tablename__ = 'users': Задает имя таблицы в базе данных — users.

• id: Уникальный идентификатор пользователя. Тип BigInteger, используется как первичный ключ.

• username: Имя пользователя (может быть пустым, так как nullable=True).

• full_name: Полное имя пользователя (также может быть пустым).

• notes: Связь «один ко многим» с таблицей Note. Позволяет пользователю иметь несколько заметок. Аргумент cascade=»all, delete‑orphan» обеспечивает автоматическое удаление всех связанных заметок, если удаляется пользователь.

Модель Note

• __tablename__ = 'notes': Имя таблицы — notes.

• id: Уникальный идентификатор заметки с автоинкрементом.

• user_id: Внешний ключ, связывающий заметку с пользователем. Указывает на id в таблице users.

• content_type: Тип содержимого заметки (например, текст, фото, видео).

• content_text: Текст заметки.

• file_id: ID файла в Telegram (если заметка содержит медиафайл).

• user: Обратная связь с моделью User. Позволяет получить пользователя, к которому принадлежит заметка.

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

Кроме того, несмотря на то что мы не описывали в этих моделях колонки updated_at и updated_at, в скором времени мы увидим, что эти поля мы получили. Это произошло из‑за того, что мы выполнили описания этих полей в базовом классе.

Описание моделей — это лишь первый шаг, но для их применения необходимо создать соответствующие таблицы в базе данных. Существует несколько способов сделать это.

На практике чаще всего используется инструмент миграций Alembic, который позволяет гибко управлять изменениями в структуре базы данных: создавать таблицы, изменять или удалять колонки. Это особенно полезно для долгосрочных проектов.

Однако сегодня мы пойдем другим путем — напрямую создадим таблицы, используя встроенные средства SQLAlchemy. Для этого мы воспользуемся следующим асинхронным методом:

async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Здесь, при помощи метода create_all, мы создаем таблицы в базе данных на основе наших моделей. Далее мы привяжем вызов этой функции к запуску бота (функция start).

Данный метод поместим в ещё один созданный файл. Назовем его base.py.

from .database import async_session, engine, Base


def connection(func):
    async def wrapper(*args, **kwargs):
        async with async_session() as session:
            return await func(session, *args, **kwargs)

    return wrapper


async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

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

Данный декоратор будет выполнять следующие функции:

• async with async_session() as session: Открывает асинхронную сессию с базой данных.

• await func(session, *args, **kwargs): Передает открытую сессию в оборачиваемую функцию, чтобы она могла использовать её для выполнения запросов.

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

И теперь напишем функции, которые позволят работать с данными базы данных. Методы я пропишу в файле dao.py.

Кода будет достаточно много, поэтому я буду стараться описывать его минимумом комментариев. Для большего понимания ознакомьтесь с официальной документацией SQLAlchemy или поищите более детальное описание методов. Со своей стороны скажу, что старался сделать функции максимально доступными каждому.

Начнем с импортов.

from create_bot import logger
from .base import connection
from .models import User, Note
from sqlalchemy import select
from typing import List, Dict, Any, Optional
from sqlalchemy.exc import SQLAlchemyError

Первый импорт logger пока можно пропустить, мы его создадим позже.

Далее мы импортируем декоратор connection и наши модели таблиц (User, Note), с которыми будем работать.

Также импортированы несколько полезных методов из SQLAlchemy, с которыми мы познакомимся в процессе дальнейшей разработки.

Напишем первый метод. Он будет будет проверять есть ли пользователь в таблице users. Если есть, то будет возвращать информацию о нем, а если нет, то будет его создавать. После этот метод мы прикрутим к обработчику команды /start в боте.

@connection
async def set_user(session, tg_id: int, username: str, full_name: str) -> Optional[User]:
    try:
        user = await session.scalar(select(User).filter_by(id=tg_id))

        if not user:
            new_user = User(id=tg_id, username=username, full_name=full_name)
            session.add(new_user)
            await session.commit()
            logger.info(f"Зарегистрировал пользователя с ID {tg_id}!")
            return None
        else:
            logger.info(f"Пользователь с ID {tg_id} найден!")
            return user
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при добавлении пользователя: {e}")
        await session.rollback()

Обратите внимание. Мы повесили наш декоратор. Он генерирует переменную session, подставляя туда значение. Остальные аргументы, такие как tg_id, username и full_name уже необходимо будет передать самостоятельно.

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

Теперь напишем метод для добавления заметки.

@connection
async def add_note(session, user_id: int, content_type: str,
                   content_text: Optional[str] = None, file_id: Optional[str] = None) -> Optional[Note]:
    try:
        user = await session.scalar(select(User).filter_by(id=user_id))
        if not user:
            logger.error(f"Пользователь с ID {user_id} не найден.")
            return None

        new_note = Note(
            user_id=user_id,
            content_type=content_type,
            content_text=content_text,
            file_id=file_id
        )

        session.add(new_note)
        await session.commit()
        logger.info(f"Заметка для пользователя с ID {user_id} успешно добавлена!")
        return new_note
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при добавлении заметки: {e}")
        await session.rollback()

Сначала проверяется, существует ли пользователь с указанным user_id в базе данных. Если пользователь найден, создается новый экземпляр модели Note, в который передаются все необходимые параметры: тип содержимого (content_type), текст заметки (content_text) и ID файла (file_id). Затем новая заметка добавляется в сессию с помощью session.add(), а изменения сохраняются вызовом session.commit().

Этот пример взаимодействия с базой данных выглядит максимально «питонично» и лаконично — именно за это я и ценю SQLAlchemy. Нам не нужно вникать в тонкости выполнения SQL-запросов. ORM берет на себя всю рутину, позволяя сосредоточиться на бизнес-логике, оставляя низкоуровневые операции на уровне абстракции.

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

@connection
async def update_text_note(session, note_id: int, content_text: str) -> Optional[Note]:
    try:
        note = await session.scalar(select(Note).filter_by(id=note_id))
        if not note:
            logger.error(f"Заметка с ID {note_id} не найдена.")
            return None

        note.content_text = content_text
        await session.commit()
        logger.info(f"Заметка с ID {note_id} успешно обновлена!")
        return note
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при обновлении заметки: {e}")
        await session.rollback()

Эта функция обновляет текстовое содержимое заметки в базе данных:

1. Проверяется наличие заметки с указанным note_id. Если заметка не найдена, логируется ошибка, и функция возвращает None.

2. Если заметка существует, обновляется её текст (content_text), после чего изменения сохраняются в базе с помощью commit().

3. В случае ошибки логируется сообщение об ошибке, и выполняется откат изменений с помощью rollback().

Функция возвращает обновленную заметку или None, если обновление не удалось.

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

Метод для получения заметки по ее ID.

@connection
async def get_note_by_id(session, note_id: int) -> Optional[Dict[str, Any]]:
    try:
        note = await session.get(Note, note_id)
        if not note:
            logger.info(f"Заметка с ID {note_id} не найдена.")
            return None

        return {
            'id': note.id,
            'content_type': note.content_type,
            'content_text': note.content_text,
            'file_id': note.file_id
        }
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при получении заметки: {e}")
        return None

Метод действительно мог бы быть проще, но я решил сделать его более гибким, возвращая результат в виде Python-словаря для удобства работы. В примере для получения записи используется метод get, а не filter_by. Основное различие в том, что get позволяет мгновенно получить запись, если известен её первичный ключ (ID), независимо от названия колонки. Это делает запросы более лаконичными и эффективными.

Теперь перейдём к описанию метода для удаления заметки по её ID. Мы просто находим запись с помощью get, затем удаляем её из сессии и сохраняем изменения вызовом session.commit():

@connection
async def delete_note_by_id(session, note_id: int) -> Optional[Note]:
    try:
        note = await session.get(Note, note_id)
        if not note:
            logger.error(f"Заметка с ID {note_id} не найдена.")
            return None

        await session.delete(note)
        await session.commit()
        logger.info(f"Заметка с ID {note_id} успешно удалена.")
        return note
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при удалении заметки: {e}")
        await session.rollback()
        return None

Принцип тут простой. Получаем заметку, а затем, если она есть, методом delete удаляем ее.

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

@connection
async def get_notes_by_user(session, user_id: int, date_add: str = None, text_search: str = None,
                            content_type: str = None) -> List[Dict[str, Any]]:
    try:
        result = await session.execute(select(Note).filter_by(user_id=user_id))
        notes = result.scalars().all()

        if not notes:
            logger.info(f"Заметки для пользователя с ID {user_id} не найдены.")
            return []

        note_list = [
            {
                'id': note.id,
                'content_type': note.content_type,
                'content_text': note.content_text,
                'file_id': note.file_id,
                'date_created': note.created_at
            } for note in notes
        ]

        if date_add:
            note_list = [note for note in note_list if note['date_created'].strftime('%Y-%m-%d') == date_add]

        if text_search:
            note_list = [note for note in note_list if text_search.lower() in (note['content_text'] or '').lower()]

        if content_type:
            note_list = [note for note in note_list if note['content_type'] == content_type]

        return note_list
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при получении заметок: {e}")
        return []

Всё начинается с того, что с помощью filter_by мы получаем все заметки, принадлежащие пользователю. Если такие заметки найдены, я преобразую их в список Python-словарей. Хотя существует множество подходов для работы с данными, я выбрал этот за его простоту и наглядность.

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

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

На этом мы завершили подготовку базы данных для бота. Теперь можно приступать к разработке его функционала!

Начинаем писать код бота

Файловая структура в этом проекте будет аналогичной той, что я использовал во всех своих предыдущих ботах, которые уже подробно описывал на Хабре. Начнём с подготовки файла с переменными окружения .env:

TOKEN=0000AABB
ADMINS=123456,4433455

Здесь у нас будет две переменные:

1. TOKEN — токен бота, который вы получили от BotFather.

2. ADMINS — список ID администраторов, разделённый запятыми.

Теперь создадим файл create_bot.py. В этом файле будут храниться настройки и переменные, которые мы будем использовать по всему боту.

import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from decouple import config


admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())

Кратко пройдемся по основным моментам:

• Переменная ADMINS трансформируется из строки в список целых чисел.

• Мы настраиваем логгер, который будет выводить информацию о работе бота (например, сообщения об ошибках и запросах).

• Далее создаём два ключевых объекта Aiogram 3:

  • Bot — объект, который взаимодействует с Telegram API. С его помощью можно отправлять и получать сообщения, работать с пользователями, чатами и выполнять различные запросы к Telegram.

  • Dispatcher — отвечает за управление событиями: регистрацией обработчиков (handlers), обработкой команд, сообщений, колбэков и других событий.

Эти два объекта являются основой для работы любого бота на базе Aiogram.

Обратите внимание. В качестве хранилища для FSM я использовал MemoryStorage. В боевых проектах лучше не использовать его, а отдавать предпочтение RedisStorage. Почему так и вообще, более подробный разбор темы FSM в aiogram 3 я давал в этой статье: "Telegram Боты на Aiogram 3.x: Все про FSM простыми словами".

Теперь опишем главный файл для запуска бота. Назовем его aiogram_run.py. Данный файл будет собирать весь наш проект в одно целое, а затем будет выполнять его запуск.

import asyncio
from create_bot import bot, dp, admins
from data_base.base import create_tables
from handlers.note.find_note_router import find_note_router
from handlers.note.upd_note_router import upd_note_router
from handlers.note.add_note_router import add_note_router
from aiogram.types import BotCommand, BotCommandScopeDefault

from handlers.start_router import start_router


# Функция, которая настроит командное меню (дефолтное для всех пользователей)
async def set_commands():
    commands = [BotCommand(command='start', description='Старт')]
    await bot.set_my_commands(commands, BotCommandScopeDefault())


# Функция, которая выполнится когда бот запустится
async def start_bot():
    await set_commands()
    await create_tables()
    for admin_id in admins:
        try:
            await bot.send_message(admin_id, f'Я запущен?.')
        except:
            pass


# Функция, которая выполнится когда бот завершит свою работу
async def stop_bot():
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, 'Бот остановлен. За что??')
    except:
        pass


async def main():
    # регистрация роутеров
    dp.include_router(start_router)
    dp.include_router(add_note_router)
    dp.include_router(find_note_router)
    dp.include_router(upd_note_router)

    # регистрация функций
    dp.startup.register(start_bot)
    dp.shutdown.register(stop_bot)

    # запуск бота в режиме long polling при запуске бот очищает все обновления, которые были за его моменты бездействия
    try:
        await bot.delete_webhook(drop_pending_updates=True)
        await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
    finally:
        await bot.session.close()


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

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

Из того на что стоит обратить внимание – это функция start_bot. При ее вызове будет монтироваться командное меню, затем будут создаваться таблицы в базе данных и после админы будут получать сообщение с текстом о том, что бот запущен.

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

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

Подготовим клавиатуры для бота

Клавиатуры я описываю в пакете keyboards. Тут у нас будет 2 файла: note_kb.py (клавиатуры, которые имеют отношение только к заметкам) и other_kb.py (универсальные клавиатуры).

Файл other_kb.py

from aiogram.types import KeyboardButton, ReplyKeyboardMarkup


def main_kb():
    kb_list = [
        [KeyboardButton(text="? Заметки")]
    ]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Воспользуйся меню?"
    )


def stop_fsm():
    kb_list = [
        [KeyboardButton(text="❌ Остановить сценарий")],
        [KeyboardButton(text="? Главное меню")]
    ]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Для того чтоб остановить сценарий FSM нажми на одну из двух кнопок?"
    )

Тут я описал две простые текстовые клавиатуры. Первая клавиатура будет отправлять клавиатуру главного меню (main_kb), вторая клавиатура (stop_fsm) будет появляться при сценариях FSM.

Файл note_kb.py

from aiogram.types import KeyboardButton, ReplyKeyboardMarkup
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton


def generate_date_keyboard(notes):
    unique_dates = {note['date_created'].strftime('%Y-%m-%d') for note in notes}
    keyboard = InlineKeyboardMarkup(inline_keyboard=[])
    for date_create in unique_dates:
        button = InlineKeyboardButton(text=date_create, callback_data=f"date_note_{date_create}")
        keyboard.inline_keyboard.append([button])

    keyboard.inline_keyboard.append([InlineKeyboardButton(text="Главное меню", callback_data="main_menu")])

    return keyboard


def generate_type_content_keyboard(notes):
    unique_content = {note['content_type'] for note in notes}
    keyboard = InlineKeyboardMarkup(inline_keyboard=[])
    for content_type in unique_content:
        button = InlineKeyboardButton(text=content_type, callback_data=f"content_type_note_{content_type}")
        keyboard.inline_keyboard.append([button])

    keyboard.inline_keyboard.append([InlineKeyboardButton(text="Главное меню", callback_data="main_menu")])

    return keyboard


def main_note_kb():
    kb_list = [
        [KeyboardButton(text="? Добавить заметку"), KeyboardButton(text="? Просмотр заметок")],
        [KeyboardButton(text="? Главное меню")]
    ]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Воспользуйся меню?"
    )


def find_note_kb():
    kb_list = [
        [KeyboardButton(text="? Все заметки"), KeyboardButton(text="? По дате добавления"), ],
        [KeyboardButton(text="? Поиск по тексту"), KeyboardButton(text="? По типу контента")],
        [KeyboardButton(text="? Главное меню")]
    ]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Выберите опцию?"
    )


def rule_note_kb(note_id: int):
    return InlineKeyboardMarkup(
        inline_keyboard=[[InlineKeyboardButton(text="Изменить текст", callback_data=f"edit_note_text_{note_id}")],
                         [InlineKeyboardButton(text="Удалить", callback_data=f"dell_note_{note_id}")]])


def add_note_check():
    kb_list = [
        [KeyboardButton(text="✅ Все верно"), KeyboardButton(text="❌ Отменить")]
    ]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Воспользуйся меню?"
    )

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

Пакет utils

Теперь создадим пакет utils и внутри создадим файл utils.py, прописав внутри этого файла дополнительные универсальные утилиты. На этом файле мы остановимся подробнее, так как понимание этих утилит позволит вам понять общие принципы работы бота.

Импорты.

import asyncio
from aiogram.types import Message
from keyboards.note_kb import rule_note_kb

asyncio я импортировал для асинхронных пауз в рассылке сообщения.

Message для аннотации объекта с которым будут работать утилиты.

И клавиатура rule_note_kb. Она будет под каждой заметкой подставлять кнопки управления заметкой: «Изменить текст» и «Удалить». Принимает эта функция ID заметки.

Теперь напишем первую функцию. Она будет принимать объект класса Message и будет возвращать питоновский словарь с такими значениями:

• content_type: это строка, содержащая одно из значений типа контента. Такие варианты: phtoto, video, text и так далее.

• file_id: это строка, которая будет содержать айди медиафайла (фото лучшего качества, документа, видео и т.д.) или None если было отправлено простое сообщение.

• content_text: тут будет храниться или текст сообщения для текстового сообщения или текст описания к медиа (caption) если есть описание. Если было отправлено медиа сообщение без комментария то будет None.

def get_content_info(message: Message):
    content_type = None
    file_id = None

    if message.photo:
        content_type = "photo"
        file_id = message.photo[-1].file_id
    elif message.video:
        content_type = "video"
        file_id = message.video.file_id
    elif message.audio:
        content_type = "audio"
        file_id = message.audio.file_id
    elif message.document:
        content_type = "document"
        file_id = message.document.file_id
    elif message.voice:
        content_type = "voice"
        file_id = message.voice.file_id
    elif message.text:
        content_type = "text"

    content_text = message.text or message.caption
    return {'content_type': content_type, 'file_id': file_id, 'content_text': content_text}

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

Следующая универсальная функция будет выполнять отправку заметки с любым типом контента.

async def send_message_user(bot, user_id, content_type, content_text=None, file_id=None, kb=None):
    if content_type == 'text':
        await bot.send_message(chat_id=user_id, text=content_text, reply_markup=kb)
    elif content_type == 'photo':
        await bot.send_photo(chat_id=user_id, photo=file_id, caption=content_text, reply_markup=kb)
    elif content_type == 'document':
        await bot.send_document(chat_id=user_id, document=file_id, caption=content_text, reply_markup=kb)
    elif content_type == 'video':
        await bot.send_video(chat_id=user_id, video=file_id, caption=content_text, reply_markup=kb)
    elif content_type == 'audio':
        await bot.send_audio(chat_id=user_id, audio=file_id, caption=content_text, reply_markup=kb)
    elif content_type == 'voice':
        await bot.send_voice(chat_id=user_id, voice=file_id, caption=content_text, reply_markup=kb)


# Улучшенная версия кода для для Python 3.10+
async def send_message_user(bot, user_id, content_type, content_text=None, file_id=None, kb=None):
    match content_type:
        case 'text': await bot.send_message(chat_id=user_id, text=content_text, reply_markup=kb)
        case 'photo': await bot.send_photo(chat_id=user_id, photo=file_id, caption=content_text, reply_markup=kb)
        case 'document': await bot.send_document(chat_id=user_id, document=file_id, caption=content_text, reply_markup=kb)
        case 'video': await bot.send_video(chat_id=user_id, video=file_id, caption=content_text, reply_markup=kb)
        case 'audio': await bot.send_audio(chat_id=user_id, audio=file_id, caption=content_text, reply_markup=kb)
        case 'voice': await bot.send_voice(chat_id=user_id, voice=file_id, caption=content_text, reply_markup=kb)

За пример оптимизированной функции под Python 3.10+ спасибо пользователю IvanZaycev0717

Все достаточно логично. Функция принимает объект бота и необходимые параметры для отправки сообщения. Обратите внимание, что данная функция также поддерживает клавиатуру, что делает ее действительно универсальной.

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

Также в утилитах я описал менее универсальную функцию для чистоты кода. Она позволяет массово отправлять заметки пользователю.

async def send_many_notes(all_notes, bot, user_id):
    for note in all_notes:
        try:
            await send_message_user(bot=bot, content_type=note['content_type'],
                                    content_text=note['content_text'],
                                    user_id=user_id,
                                    file_id=note['file_id'],
                                    kb=rule_note_kb(note['id']))
        except Exception as E:
            print(f'Error: {E}')
            await asyncio.sleep(2)
        finally:
            await asyncio.sleep(0.5)

Пишем хендлеры бота

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

Для написания хендлеров подготовим пакет handlers и там создадим файл start_router.py.

Начнем с импортов.

from aiogram import Router, F
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery
from data_base.dao import set_user
from keyboards.other_kb import main_kb

FSMContext я импортировал для того чтоб в стартовых методах очищать хранилище машины состоянии. Это бывает очень полезно и в статье про FSM я подробно описывал почему.

Теперь начнем писать код первого хендлера.

start_router = Router()


# Хендлер команды /start и кнопки "? Главное меню"
@start_router.message(F.text == '? Главное меню')
@start_router.message(CommandStart())
async def cmd_start(message: Message, state: FSMContext):
    await state.clear()
    user = await set_user(tg_id=message.from_user.id,
                          username=message.from_user.username,
                          full_name=message.from_user.full_name)
    greeting = f"Привет, {message.from_user.full_name}! Выбери необходимое действие"
    if user is None:
        greeting = f"Привет, новый пользователь! Выбери необходимое действие"

    await message.answer(greeting, reply_markup=main_kb())

Тут мы создали стартовый роутер, который будет выступать тут в роли объекта диспетчер.

Вход в главное меню я прописал двумя декораторами.

@start_router.message(F.text == '? Главное меню')

Этот декоратор будет срабатывать не текстовое сообщение '? Главное меню'

@start_router.message(CommandStart())

Этот декоратор будет срабатывать на команду /start. Использовал для этого встроенные фильтры aiogram 3, а именно CommandStart().

Теперь по самому коду.

В начале мы очищаем хранилище FSM – await state.clear()

И далее уже идет взаимодействие с базой данных через подготовленную ранее функцию. Благодаря логике, которую мы заложили ранее мы или зарегистрируем пользователя или просто вернем по нему информацию. Далее мы отправим пользователю сообщение с клавиатурой главного меню.

В этом же файле я описал ещё две функции.

@start_router.message(F.text == '❌ Остановить сценарий')
async def stop_fsm(message: Message, state: FSMContext):
    await state.clear()
    await message.answer(f"Сценарий остановлен. Для выбора действия воспользуйся клавиатурой ниже",
                         reply_markup=main_kb())


@start_router.callback_query(F.data == 'main_menu')
async def main_menu_process(call: CallbackQuery, state: FSMContext):
    await state.clear()
    await call.answer('Вы вернулись в главное меню.')
    await call.message.answer(f"Привет, {call.from_user.full_name}! Выбери необходимое действие",
                              reply_markup=main_kb())

Первая функция будет останавливать сценарий FSM, независимо от места сценария, в котором этот кусок кода был вызван. В отличие от Aiogram 2, в тройке state=[“*”] установлен по умолчанию.

Вторая функция выполняет ту же логику, но уже в контексте calback, а не message.

Теперь мы готовы выполнить первый запуск бота. Если все корректно, то будет создана база данных в корне проекта бота и я, как администратор, получу сообщение о том, что бот запущен. Проверим.

Запускаем файл aiogram_run.py

Вижу, что ошибок после запуска нет и бот сообщил мне о том, что он запущен.

Выполню в боте команду /start

Обратите внимание, что после повторного вызова /start текст сообщения от бота изменился и это значит, что мой телеграмм ID попал в базу данных. Проверим.

Я вижу, что моя запись в базе данных и что существует две таблицы. Это значит, что пока все работает корректно. Теперь напишем код для работы с заметками.

Пишем код для работы с заметками

Напоминаю, что у нас будет функционал для: добавления, просмотра и редактирования заметок. Для удобства внутри пакета handlers я создал пакет note и там каждое это действие разбил на отдельные файлы.

Файл add_note_router.py

Как вы поняли из названия, в этом файле мы опишем логику добавления нашей заметки. Начнем с импортов:

from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message
from create_bot import bot
from data_base.dao import add_note
from keyboards.note_kb import main_note_kb, add_note_check
from keyboards.other_kb import stop_fsm
from utils.utils import get_content_info, send_message_user

Тут появился следующий импорт from aiogram.fsm.state import StatesGroup, State, что намекает на то что мы будем использовать машину состояний.

По остальным импортам, думаю, все понятно. Тут у нас метод для добавления заметки в базу данных, клавиатура и методы с файла utils.py.

Подготовим класс для работы с состояниями.

class AddNoteStates(StatesGroup):
    content = State()  # Ожидаем любое сообщение от пользователя
    check_state = State()  # Финальна проверка

Нас будет интересовать всего два состояния: когда пользователь отправил свое сообщение и когда он нажал на клавишу проверки.

Отправим главное сообщение после входа в блок заметок.

@add_note_router.message(F.text == '? Заметки')
async def start_note(message: Message, state: FSMContext):
    await state.clear()
    await message.answer('Ты в меню добавления заметок. Выбери необходимое действие.',
                         reply_markup=main_note_kb())

Тут все просто. Мы очищаем состояние и отправляем сообщение с действиями с заметками.

Теперь добавим сценарий, который пойдет после клика на кнопку «Добавить заметку».

@add_note_router.message(F.text == '? Добавить заметку')
async def start_add_note(message: Message, state: FSMContext):
    await state.clear()
    await message.answer('Отправь сообщение в любом формате (текст, медиа или медиа + текст). '
                         'В случае если к медиа требуется подпись - оставь ее в комментариях к медиа-файлу ',
                         reply_markup=stop_fsm())
    await state.set_state(AddNoteStates.content)

На данном этапе мы переводим пользователя в состояние ожидания сообщения и даем возможность выйти с этого состояния нажав на кнопку «Главное меню» или «Остановить сценарий».

Как раз для того чтоб была возможность выходить с состояний ожиданий мы прописали state.clear() во всех хендлерах.

Теперь напишем обработчик входящего сообщения от пользователя.

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

В начале мы получаем словарь, описывающий входящее сообщение.

content_info = get_content_info(message)

Далее, на основании этих данных, мы сформировали сообщение с проверкой, которое мы отправили при помощи другого, заранее подготовленного метода.

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

Напишем обработчики для «Все верно» и «Отменить»

@add_note_router.message(AddNoteStates.check_state, F.text == '✅ Все верно')
async def confirm_add_note(message: Message, state: FSMContext):
    note = await state.get_data()
    await add_note(user_id=message.from_user.id, content_type=note.get('content_type'),
                   content_text=note.get('content_text'), file_id=note.get('file_id'))
    await message.answer('Заметка успешно добавлена!', reply_markup=main_note_kb())
    await state.clear()


@add_note_router.message(AddNoteStates.check_state, F.text == '❌ Отменить')
async def cancel_add_note(message: Message, state: FSMContext):
    await message.answer('Добавление заметки отменено!', reply_markup=main_note_kb())
    await state.clear()

Тут все просто. Или мы выполняем сохранение полученных данных в базу или очищаем хранилище.

Я добавлю несколько заметок разного типа данных.

Теперь опишем логику отображения / поиска заметок. Для этого я создам файл find_note_router.py.

Импорты

from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery
from create_bot import bot
from data_base.dao import get_notes_by_user
from keyboards.note_kb import main_note_kb, find_note_kb, generate_date_keyboard, generate_type_content_keyboard
from utils.utils import send_many_notes

Импортов больше, но общую логику вы понимаете. Метод для фильтрации и получения заметок, клавиатуры.

Для хранения состояний нам достаточно будет одного класса:

class FindNoteStates(StatesGroup):
    text = State()  # Ожидаем текст для поиска заметок

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

Метод входа в сценарий.

@find_note_router.message(F.text == '? Просмотр заметок')
async def start_views_noti(message: Message, state: FSMContext):
    await state.clear()
    await message.answer('Выбери какие заметки отобразить', reply_markup=find_note_kb())

Тут запускается клавиатура с вариантами поиска заметок. Начнем с самого простого фильтра – «Все заметки».

@find_note_router.message(F.text == '? Все заметки')
async def all_views_noti(message: Message, state: FSMContext):
    await state.clear()
    all_notes = await get_notes_by_user(user_id=message.from_user.id)
    if all_notes:
        await send_many_notes(all_notes, bot, message.from_user.id)
        await message.answer(f'Все ваши {len(all_notes)} заметок отправлены!', reply_markup=main_note_kb())
    else:
        await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())

Тут мы просто вызываем метод get_notes_by_user, передавая в него user_id. Сам user_id мы берем из message. Далее, при помощи подготовленной функции с утилит выполняем массовую отправку заметок с инлайн клавиатурой для их редактирования.

Все заметки отобразились.

Поиск заметок по дате добавления.

@find_note_router.message(F.text == '? По дате добавления')
async def date_views_noti(message: Message, state: FSMContext):
    await state.clear()
    all_notes = await get_notes_by_user(user_id=message.from_user.id)
    if all_notes:
        await message.answer('На какой день вам отобразить заметки?',
                             reply_markup=generate_date_keyboard(all_notes))
    else:
        await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())


@find_note_router.callback_query(F.data.startswith('date_note_'))
async def find_note_to_date(call: CallbackQuery, state: FSMContext):
    await call.answer()
    await state.clear()
    date_add = call.data.replace('date_note_', '')
    all_notes = await get_notes_by_user(user_id=call.from_user.id, date_add=date_add)
    await send_many_notes(all_notes, bot, call.from_user.id)
    await call.message.answer(f'Все ваши {len(all_notes)} заметок на {date_add} отправлены!',
                              reply_markup=main_note_kb())

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

Вот как выглядит генерация этой клавиатуры.

def generate_date_keyboard(notes):
    unique_dates = {note['date_created'].strftime('%Y-%m-%d') for note in notes}
    keyboard = InlineKeyboardMarkup(inline_keyboard=[])
    for date_create in unique_dates:
        button = InlineKeyboardButton(text=date_create, callback_data=f"date_note_{date_create}")
        keyboard.inline_keyboard.append([button])

    keyboard.inline_keyboard.append([InlineKeyboardButton(text="Главное меню", callback_data="main_menu")])

    return keyboard

Для наглядности я подправил дату добавления одной заметки и вот что у меня получилось.

Вызову заметку на 2024-06-21

Фильтрация по типу контента работает точно по такому же принципу, только сбор идет не по колонке с датой публикации, а по колонке с типом контента.

Функция для генерации клавиатуры.

def generate_type_content_keyboard(notes):
    unique_content = {note['content_type'] for note in notes}
    keyboard = InlineKeyboardMarkup(inline_keyboard=[])
    for content_type in unique_content:
        button = InlineKeyboardButton(text=content_type, callback_data=f"content_type_note_{content_type}")
        keyboard.inline_keyboard.append([button])

    keyboard.inline_keyboard.append([InlineKeyboardButton(text="Главное меню", callback_data="main_menu")])

    return keyboard

И сама логика поиска по типу контента.

@find_note_router.message(F.text == '? По типу контента')
async def content_type_views_noti(message: Message, state: FSMContext):
    await state.clear()
    all_notes = await get_notes_by_user(user_id=message.from_user.id)
    if all_notes:
        await message.answer('Какой тип заметок по контенту вас интересует?',
                             reply_markup=generate_type_content_keyboard(all_notes))
    else:
        await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())


@find_note_router.callback_query(F.data.startswith('content_type_note_'))
async def find_note_to_content_type(call: CallbackQuery, state: FSMContext):
    await call.answer()
    await state.clear()
    content_type = call.data.replace('content_type_note_', '')
    all_notes = await get_notes_by_user(user_id=call.from_user.id, content_type=content_type)
    await send_many_notes(all_notes, bot, call.from_user.id)
    await call.message.answer(f'Все ваши {len(all_notes)} с типом контента {content_type} отправлены!',
                              reply_markup=main_note_kb())

Поиск по текстовому содержимому будет немного отличаться.

@find_note_router.message(F.text == '? Поиск по тексту')
async def text_views_noti(message: Message, state: FSMContext):
    await state.clear()
    all_notes = await get_notes_by_user(user_id=message.from_user.id)
    if all_notes:
        await message.answer('Введите поисковой запрос. После этого я начну поиск по заметкам. Если в текстовом '
                             'содержимом заметки будет обнаружен поисковой запрос, то я отображу эти заметки')
        await state.set_state(FindNoteStates.text)
    else:
        await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())


@find_note_router.message(F.text, FindNoteStates.text)
async def text_noti_process(message: Message, state: FSMContext):
    text_search = message.text.strip()
    all_notes = await get_notes_by_user(user_id=message.from_user.id, text_search=text_search)
    await state.clear()
    if all_notes:
        await send_many_notes(all_notes, bot, message.from_user.id)
        await message.answer(f'C поисковой фразой {text_search} было обнаружено {len(all_notes)} заметок!',
                             reply_markup=main_note_kb())
    else:
        await message.answer(f'У вас пока нет ни одной заметки, которая содержала бы в тексте {text_search}!',
                             reply_markup=main_note_kb())

Обратите внимание, тут сработало игнорирование регистра поискового запроса.

Таким образом мы закрыли вопрос поиска и добавления заметок и нам осталось решить вопрос с изменением / удалением заметок.

Для этой задачи я подготовил файл upd_note_router.py

Тут нам нужно будет хранить новое текстовое содержимое для заметки.

class UPDNoteStates(StatesGroup):
    content_text = State()

Реализуем логику для изменения текстового содержимого в заметке.

Для входа в этот режим используется inline клавиатура с call_data = f«edit_note_text_{note_id}».

@upd_note_router.callback_query(F.data.startswith('edit_note_text_'))
async def edit_note_text_process(call: CallbackQuery, state: FSMContext):
    await state.clear()
    note_id = int(call.data.replace('edit_note_text_', ''))
    await call.answer(f'Режим редактирования заметки с ID {note_id}')
    await state.update_data(note_id=note_id)
    await call.message.answer(f'Отправь новое текстовое содержимоем для заметки с ID {note_id}')
    await state.set_state(UPDNoteStates.content_text)

Этой логикой мы запустили ожидание нового текстового содержимого, предварительно извлекая note_id из call_data. Подробнее о том, как работают инлайн-клавиатуры писал в этой статье.

Далее нам остается перезаписать текстовое содержимое у заметки с ID note_id.

@upd_note_router.message(F.text, UPDNoteStates.content_text)
async def confirm_edit_note_text(message: Message, state: FSMContext):
    note_data = await state.get_data()
    note_id = note_data.get('note_id')
    content_text = message.text.strip()
    await update_text_note(note_id=note_id, content_text=content_text)
    await state.clear()
    await message.answer(f'Текст заметки с ID {note_id} успешно изменен на {content_text}!',
                         reply_markup=main_note_kb())

И последнее. Опишем логику для удаления заметки.

@upd_note_router.callback_query(F.data.startswith('dell_note_'))
async def dell_note_process(call: CallbackQuery, state: FSMContext):
    await state.clear()
    note_id = int(call.data.replace('dell_note_', ''))
    await delete_note_by_id(note_id=note_id)
    await call.answer(f'Заметка с ID {note_id} удалена!', show_alert=True)
    await call.message.delete()

Бот полностью готов и теперь нам остается последний штрих – выполнить удаленный запуск бота в облаке (деплой).

Подготовка к деплою бота

Для начала, в корне проекта, необходимо создать файл с именем amvera.yml со следующим содержимым:

meta:
  environment: python
  toolchain:
    name: pip
    version: "3.12"
build:
  requirementsPath: requirements.txt
run:
  scriptName: aiogram_run.py

Этими простыми настройками мы указываем, что работать будем с Python 3.12, для установки будем использовать pip. Кроме того, в этом файле необходимо указать путь к файлу requirements.txt и имя файла запуска бота.

Проверьте, чтоб перед деплоем у вас была такая структура проекта.

На этом подготовка завершена, и мы можем переходить к деплою бота на сервис Amvera Cloud.

Деплой telegram-бота

Следуйте этим простым шагам и уже через пару минут ваш бот будет запущен на удаленном хостинге.

• Выполняем регистрацию на сервисе Amvera Cloud (новые пользователи на баланс получают 111 рублей)

• Переходим в раздел проектов

• Создаем новый проект. На этом этапе нужно придумать имя проекту и выбрать тариф.

• На открывшемся экране необходимо выбрать «Через интерфейс» и загрузить файлы бота с файлом amvera.yml. Затем жмем на «Далее».

• На следующем экране должны отобразиться ваши настройки. Проверьте, что все введено корректно и жмем на «Завершить»

После этих простых действий остается подождать 2-3 минуты. В это время проект с ботом сначала соберется, а после, Amvera запустит этот проект.

Заключение

Друзья, этот бот — учебный проект, а не законченный инструмент для работы с заметками в Telegram. Моя цель была не создать идеальный продукт, а показать ключевые принципы разработки. В некоторых местах я намеренно упростил код. Например, в реальных проектах обычно применяют PostgreSQL для базы данных, Alembic для управления миграциями схем и Redis для работы с машиной состояний. Также в SQLAlchemy можно реализовать более эффективные решения — индексы, связи между таблицами и многое другое.

Но, повторюсь, главная задача этого проекта — обучить. Сегодня мы рассмотрели основные принципы интеграции ботов в Telegram с использованием SQLAlchemy, научились работать с медиафайлами, сохранять их в облаке, а также узнали, как организовать поиск по базе данных и многое другое.

Если вас заинтересовала эта тема и будет поддержка, я готов продолжать развивать проект. У меня есть идеи для расширения функционала заметок, например, добавление тегов, а также блок задач и напоминаний. Всё это будет зависеть от вашей активности. Не забудьте проголосовать под статьёй, если хотите видеть цикл моих публикаций о работе с SQLAlchemy!

Полный исходный код бота, а также эксклюзивные материалы, которые я не выкладываю на Хабре, доступны в моём Telegram-канале «Легкий путь в Python».

До новых встреч!