19 октября

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

PythonЯндекс APIМикросервисы

Вводная

Данная статья является продолжением вот этой статьи. В ней мы рассмотрели создание и настройку 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, особенно их администрации, которая очень толково все разъясняет.

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

Теги:pythontelegrambotyandex.cloudserverlessfunctions
Хабы: Python Яндекс API Микросервисы
+6
1,6k 30
Комментировать
Похожие публикации
Python Developer
от 80 000 до 200 000 ₽kt.teamМожно удаленно
Python/Django разработчик
от 1 900 до 2 400 $FloshipМожно удаленно
Python developer
от 170 000 до 200 000 ₽AIR ProductionМосква
Python Разработчик (Python Backend Developer)
от 150 000 ₽Правое полушарие ИнтровертаМожно удаленно
Python разработчик
от 70 000 до 130 000 ₽AGORAМожно удаленно
Лучшие публикации за сутки