Подключение Yandex Database к serverless телеграм боту на Yandex Functions

Вводная

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

В качестве базы данных мы будем использовать Yandex Cloud Database.

Задачи

1. Создать базу данных;

2. Подготовить базу к подключению;

3. Установить зависимости;

4. Добавить таблицу в базе данных для хранения пользователя;

5. Сохранить информацию о входящем пользователе в сообщении телеграмма;

6. Получить информацию и отправить её пользователю из базы данных.

Создание базы данных

Самая простая задача в нашем списке, нам нужно зайти в Консоль Yandex Cloud под своим аккаунтом. Затем в меню консоли управления выбираем Yandex Database.

Где найти кнопку

Кликаем на кнопку . Здесь мы можем задать имя базы и тип. В качестве типа я рекомендую выбирать Serverless, поскольку трафика у нас крайне мало и данных мы хранить особо много не будем. Молодцы! Мы создали базу данных.

Настройка подключения к базе данных

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

1. Создание сервисного аккаунта и получение ключей для доступов к базе;

2. Установка зависимостей для python (boto3);

3. Настройка подключения к базе данных в коде.

Для создания сервисного аккаунта (это такая сущность, которая позволяет программам получать доступ к различным функциям Яндекс Облака) читаем ВНИМАТЕЛЬНО вводную, затем заходим в свою консоль и выбираем облако, в которой расположена база данных. В нем мы нажимаем на «Сервисные аккаунты».

Где найти сервисные аккаунты

Далее создаем сервисный аккаунт с РОЛЬЮ editor. Она позволит работать нам от имени программы с нашей базой.

Как в итоге должно быть

Далее нажимаем на созданный аккаунт и находим кнопку «Создать новый ключ» и в выпадающем списке выбираем «статический ключ». Оба ключика куда-нибудь записываем. Данные ключи позволят нам настроить подключение по DocAPI к Yandex Cloud Database.

Установка зависимостей

Для установки зависимостей (библиотека для подключения к Yandex Database), мы должны перейти в Редактор нашей функции и посмотреть параметр — «Среда выполнения». Ставим 3.7 preview (так как она поддерживает установку зависимостей).

Проверяем среду выполнения

Затем создаем файл ‘requirements.txt’, он будет использоваться для хранения наших зависимостей и они станут доступны внутри функции. Внутрь пишем библиотеку boto3, это SDK для работы с AWS, но она позволит нам подключиться к Yandex Database как DynamoDB. В итоге у нас получилось 2 файла — бот и зависимости.

Два файла телеграм бота

Теперь мы умеем ставить различные библиотеки!

Прописываем подключение в коде

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

import json import logging import os  import boto3 from botocore.exceptions import ClientError  def read_user(user_id, dynamodb=None):     if not dynamodb:         dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )     table = dynamodb.Table('Users')     try:         response = table.get_item(Key={'user_id': str(user_id)})     except ClientError as e:         print(e.response['Error']['Message'])     else:         return response  def create_user(user_id, first_name, dynamodb=None):     if not dynamodb:         dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )      table = dynamodb.Table('Users')     response = table.put_item(         Item={         'user_id': str(user_id),         'first_name': str(first_name)         }     )     return response  def handler(event, context):     dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )     body = json.loads(event['body'])     user_query = read_user(body['message']['chat']['id'], dynamodb)     if 'Item' not in user_query:         create_user(body['message']['chat']['id'], body['message']['from']['first_name'], dynamodb)         return {             'statusCode': 200,             'headers': {                 'Content-Type': 'application/json'             },             'body': json.dumps({                 'method': 'sendMessage',                 'chat_id': body['message']['chat']['id'],                 'text':  'Привет! Я тебя запомнил :)'             }),             'isBase64Encoded': False         }     user = user_query['Item']     return {         'statusCode': 200,         'headers': {             'Content-Type': 'application/json'         },         'body': json.dumps({             'method': 'sendMessage',             'chat_id': body['message']['chat']['id'],             'text':  f'Привет, {user["first_name"]}!'         }),         'isBase64Encoded': False     }

Перед написанием кода создадим 3 записи в переменные окружения функции.

Оба KEY у нас беруться из сервисного аккаунта, который мы создали ранее, а ссылку на хранилище мы берем из базы данных (Document API).

Где лежит ссылка на базу данных

Сам ресурс подключается командой:

dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )

Здесь мы указываем какой ресурс из boto3 мы используем и его настройки. Тут нас интересуют endpoint_url — адрес базы данных, и два ключа — сервисного аккаунта.

Ура, теперь мы подключили базу данных в программе!

Создание таблицы в базе

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

import os  import boto3   def create_user_table():     dynamodb = boto3.resource(         'dynamodb',         endpoint_url=USER_STORAGE_URL,         region_name = 'us-east-1',         aws_access_key_id = AWS_ACCESS_KEY_ID,         aws_secret_access_key = AWS_SECRET_ACCESS_KEY         )     table = dynamodb.create_table(         TableName = 'Users',         KeySchema=[             {                 'AttributeName': 'user_id',                 'KeyType': 'HASH' # Partition key             }         ],         AttributeDefinitions=[             {'AttributeName': 'user_id', 'AttributeType': 'S'}         ]     )     return table  create_user_table()

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

В данном коде нас интересует функция dynamodb.create_table. Здесь мы указываем при создании имя таблицы(TableName), ключевые элементы (KeySchema) и тип этих элементов (AttributeDefinitions). Не забываем исполнить код. Далее мы будем работать с этой таблицей.

Проверка пользователя и создание в случае отсутствия

Создаем функцию для чтения пользователя внутри файла main.py нашего бота:

def read_user(user_id, dynamodb=None):     if not dynamodb:         dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )     table = dynamodb.Table('Users')     try:         response = table.get_item(Key={'user_id': str(user_id)})     except ClientError as e:         print(e.response['Error']['Message'])     else:         return response

Она принимает user_id (в нашем случае это id телеграм чата с ботом) и ресурс из основной функции (опционально).

Для создания пользователя также создадим функцию, которая принимает user_id и first_name телеграм юзеров, затем сохраняет их в табличке или перезаписывает старого:

def create_user(user_id, first_name, dynamodb=None):     if not dynamodb:         dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )      table = dynamodb.Table('Users')     response = table.put_item(         Item={         'user_id': str(user_id),         'first_name': str(first_name)         }     )     return response

Затем мы должны использовать эти функции внутри нашего кода корневой функции:

def handler(event, context):     dynamodb = boto3.resource(                 'dynamodb',                 endpoint_url=os.environ.get('USER_STORAGE_URL'),                 region_name = 'us-east-1',                 aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID'),                 aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY')                 )     body = json.loads(event['body'])     user_query = read_user(body['message']['chat']['id'], dynamodb)     if 'Item' not in user_query:         create_user(body['message']['chat']['id'], body['message']['from']['first_name'], dynamodb)         return {             'statusCode': 200,             'headers': {                 'Content-Type': 'application/json'             },             'body': json.dumps({                 'method': 'sendMessage',                 'chat_id': body['message']['chat']['id'],                 'text':  'Привет! Я тебя запомнил :)'             }),             'isBase64Encoded': False         }     user = user_query['Item']     return {         'statusCode': 200,         'headers': {             'Content-Type': 'application/json'         },         'body': json.dumps({             'method': 'sendMessage',             'chat_id': body['message']['chat']['id'],             'text':  f'Привет, {user["first_name"]}!'         }),         'isBase64Encoded': False     }

Обращаем внимание на 10 и 12 строку. В 10 строке мы делаем запрос на пользователя, затем в 11 проверяем наличие его. В случае отсутствия в 12 строке идет создание пользователя.

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

Спасибо телеграм каналам Яндекс Облака и Yandex Cloud Functions, особенно их администрации, которая очень толково все разъясняет.

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