🤹🏽 🏯 🚦 YandexデータベースをYandexFunctionsのサーバーレステレグラムボットに接続する 💅🏼 🚵🏽 🛌

入門

この記事では、の続きです。この記事。その中で、テレグラムボットのyandexクラウド機能の作成と構成を調べました。そして今日は、ボットのテレグラムをデータベースに接続し、ボットが通信するユーザーに関する情報を保存することを検討します。

Yandex CloudDatabaseをデータベースとして使用します。

タスク

データベースを作成します。
接続用のベースを準備します。
依存関係をインストールします。
ユーザーを格納するためのテーブルをデータベースに追加します。
着信ユーザーに関する情報をテレグラムメッセージに保存します。
情報を取得し、データベースからユーザーに送信します。

データベースの作成

リストで最も簡単なタスクは、アカウントでYandex CloudConsoleに移動する必要があります。次に、コントロールコンソールメニューで[Yandexデータベース]を選択します。

ボタンの場所

ボタンをクリックします。ここで、ベース名とタイプを設定できます。タイプとして、サーバーレスを選択することをお勧めします。トラフィックが非常に少なく、多くのデータを保存しないためです。よくやった！データベースを作成しました。

データベース接続の設定

データベースに接続するには、独自のタスクリストを作成する必要があります。

サービスアカウントを作成し、データベースにアクセスするためのキーを取得します。
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, , .

次のステップでは、メニューを開発し、注文を簡単に行えるアプリケーションをすでに実装する予定です。

YandexデータベースをYandexFunctionsのサーバーレステレグラムボットに接続する

入門

タスク

データベースの作成

データベース接続の設定

More articles: