Подключение 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, особенно их администрации, которая очень толково все разъясняет.

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

    Комментарии 0

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

    Самое читаемое