Как сохранить данные своего телеграм-бота

Продолжаем разговор, начатый в прошлой заметке про использование бесплатного хостинга Deta. Сегодня рассмотрим механизмы хранения данных применительно к телеграм ботам на aiogram 3.

Что будем использовать #

Deta предоставляет два базовых механизма для хранения данных:

Base - NoSQL база данных
Drive - файловое хранилище

И тех, и других можно иметь произвольное количество в каждом проекте. Для доступа к ним существует два механизма - браузерный UI и API для программного доступа. Так же для удобства работы с API можно воспользоваться классами из питонного пакета deta. Дока. Сегодняшняя статья будет посвящена работе с БД.

Минимальный пример #

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

Ставим пакет для работы с API: pip install deta.
Добываем ключ доступа для проекта:

Открываем в браузере список своих проектов на https://deta.space/
Выбираем проект, для которого будем создавать БД
Нажимаем в свойствах Open in Builder
В верхних вкладках выбираем Develop, и ниже Data
В появившемся окне тыкаем кнопку Data Keys
И наконец, Create new data key, спросив имя для ключа, даст нам заветную строчку

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

Вооружившись ключом и установленным пакетом, перейдём к коду:

1import deta
2
3key = "..." 
4
5deta = deta.Deta(key)
6base = deta.Base("TestName")
7base.put(123, key="SomeKey")
8print(base.get("SomeKey"))

Ожидаемый вывод кода выше:

{'key': 'SomeKey', 'value': 123}

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

Теоретический минимум #

С базой можно взаимодействовать с помощью следующих методов:

base.put - позволяет записать в базу типы dict, list, str, int, float, bool по ключу str. Так же можно задать expire_time, тогда запись будет удалена из базы по истечении заданного времени.
base.get - получает запись из базы по ключу, возвращает dict, состоящий из key и value, как видно из примера.
base.delete - удаляет запись
base.insert - то же что put, но в случае если такой ключ в базе уже есть кинеть эксепшн.
base.put_many - то же что put, но позволяет записать сразу пачку элементов.
base.update - позволяет обновить часть записи, хранящей dict
base.fetch - получает список всех значений из базы с возможностью задания фильтра

Ближе к ботам #

Самое необходимое применение механизмов, описанных выше, это создание хранилища состояния для бота:

 1from typing import Any, Optional, cast
 2
 3from aiogram import Bot
 4from aiogram.fsm.state import State
 5from aiogram.fsm.storage.base import BaseStorage, StateType, StorageKey
 6from deta import Deta
 7
 8
 9class DetaStateStorage(BaseStorage):
10    def __init__(self, deta_project_key: str):
11        self.deta_project_key = deta_project_key
12        self.deta = Deta(self.deta_project_key)
13        self.state_db = self.deta.Base("aiogram_state")
14        self.data_db = self.deta.Base("aiogram_data")
15
16    async def set_state(
17        self, bot: 'Bot', key: StorageKey, state: StateType = None,  # noqa: ARG002
18    ) -> None:
19        value = cast(str, state.state if isinstance(state, State) else state)
20        key = str(key.user_id)
21        self.state_db.put(data=value, key=key)
22
23    async def get_state(self, bot: 'Bot', key: StorageKey) -> Optional[str]:  # noqa: ARG002
24        key = str(key.user_id)
25        data = self.state_db.get(key)
26        if data is None:
27            return None
28        value = data["value"]
29        return cast(Optional[str], value)
30
31    async def set_data(self, bot: 'Bot', key: StorageKey, data: dict[str, 'Any']) -> None:  # noqa: ARG002
32        key = str(key.user_id)
33        self.data_db.put(data=data, key=key)
34
35    async def get_data(self, bot: 'Bot', key: StorageKey) -> dict[str, 'Any']:  # noqa: ARG002
36        key = str(key.user_id)
37        return self.data_db.get(key) or {}
38
39    async def close(self) -> None:
40        pass

Этот класс реализует минимальный необходимый интерфейс BaseStorage для хранилища aiogram. Он подключается при создании объекта диспетчера как именованный аргумент:

1deta_key = "..."
2storage = DetaStateStorage(deta_key)
3dispatcher = Dispatcher(storage=storage)

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

После этого в хедлерах бота появляется возможность сохранять данные и стейт с помощью аргумента state. Выглядит это следующим образом (хэндлеры написаны на основе примера из предыдущей статьи):

 1@router.message(Command(commands=["start"]))
 2async def start_handler(message: types.Message, state: FSMContext):
 3    await state.set_data({"qwe": 123})
 4    await message.answer("Data set")
 5
 6
 7@router.message(Command(commands=["get"]))
 8async def start_handler(message: types.Message, state: FSMContext):
 9    data = await state.get_data()
10    await message.answer(data["qwe"])

Взаимодействие с таким ботом выглядит так:

>>Me
/start

>>DevBot
Data set

>>Me
/get

>>DevBot
123

Аналогичным образом работает задание стейта для диалога. За деталями работы этой подсистемы не связанными с Deta можно обратиться к документации aiogram 3.

Итоги #

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

Время отклика порядка 10 ms
Максимальный размер одной записи - 400 Kb
Общий размер хранимых данных - 10 Gb

Для небольших проектов и прототипов этого хватало с запасом.

Какие ещё полезные модули можно добавить в бота, используя эти механизмы:

Хранение пришедших апдейтов для отладки
Запись логов
Кэширование медиа-файлов
Диалоговые деревья

Use it wisely!🖖