Connecting Yandex Database to a serverless telegram bot on Yandex Functions

Introductory

This article is a continuation of this article . In it, we examined the creation and configuration of the yandex cloud functions of the telegram bot. And today we will consider connecting the bot's telegrams to the database and saving any information about the user with whom the bot communicates.

We will use Yandex Cloud Database as a database .

Tasks

1. Create a database;

2. Prepare the base for connection;

3. Install dependencies;

4. Add a table to the database to store the user;

5. Save information about the incoming user in a telegram message;

6. Get information and send it to the user from the database.

Database creation

The simplest task on our list, we need to go to the Yandex Cloud Console under our account. Then select Yandex Database in the control console menu.

Where to find the button

Click on the button . Here we can set the base name and type. As a type, I recommend choosing Serverless, since we have very little traffic and we will not store much data. Well done! We have created a database.

Setting up a database connection

To connect the database, we need to create our own list of tasks:

1. Creating a service account and obtaining keys for accessing the database;

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, , .

The next step I plan to develop a menu and already implement an application in which it will be possible to simply place any orders.