👩🏿‍🎤 🙅 🎊 Connexion de la base de données Yandex à un robot de télégramme sans serveur sur les fonctions Yandex 📳 💱 ⛏️

Introduction

Cet article est une suite de cet article . Dans ce document, nous avons examiné la création et la configuration des fonctions cloud yandex du robot de télégramme. Et aujourd'hui, nous envisagerons de connecter les télégrammes du bot à la base de données et de stocker toutes les informations sur l'utilisateur avec lequel le bot communique.

Nous utiliserons Yandex Cloud Database comme base de données .

Tâches

Créez une base de données;
Préparez la base pour la connexion;
Installer les dépendances;
Ajoutez une table à la base de données pour stocker l'utilisateur;
Enregistrer les informations sur l'utilisateur entrant dans un télégramme;
Obtenez des informations et envoyez-les à l'utilisateur à partir de la base de données.

Création de base de données

La tâche la plus simple de notre liste, nous devons accéder à la Yandex Cloud Console sous notre compte. Ensuite, sélectionnez Yandex Database dans le menu de la console de contrôle.

Où trouver le bouton

Cliquez sur le bouton . Ici, nous pouvons définir le nom et le type de base. En tant que type, je recommande de choisir Serverless, car nous avons très peu de trafic et nous ne stockerons pas beaucoup de données. Bien joué! Nous avons créé une base de données.

Configurer une connexion à une base de données

Pour connecter la base de données, nous devons créer notre propre liste de tâches:

Créer un compte de service et obtenir des clés pour accéder à la base de données;
python (boto3);
.

( , ) , , . " ".

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

La prochaine étape je prévois de développer un menu et déjà implémenter une application dans laquelle il sera possible de passer simplement des commandes.

Connexion de la base de données Yandex à un robot de télégramme sans serveur sur les fonctions Yandex

Introduction

Tâches

Création de base de données

Configurer une connexion à une base de données

More articles: