توصيل قاعدة بيانات Yandex ببوت برقية بدون خادم على وظائف Yandex

استهلالي

هذه المقالة هي استمرار لهذه المقالة . في ذلك ، درسنا إنشاء وتكوين وظائف سحابة yandex لبوت telegram. واليوم سننظر في توصيل برقيات الروبوت بقاعدة البيانات وحفظ أي معلومات حول المستخدم الذي يتواصل معه الروبوت.

سنستخدم Yandex Cloud Database كقاعدة بيانات .

مهام

1. إنشاء قاعدة بيانات ؛

2. تحضير القاعدة للاتصال ؛

3. تثبيت التبعيات

4. إضافة جدول إلى قاعدة البيانات لتخزين المستخدم ؛

5. حفظ المعلومات حول المستخدم الوارد في رسالة برقية ؛

6. الحصول على المعلومات وإرسالها إلى المستخدم من قاعدة البيانات.

إنشاء قاعدة البيانات

أبسط مهمة في قائمتنا ، نحتاج إلى الانتقال إلى Yandex Cloud Console ضمن حسابنا. ثم حدد قاعدة بيانات Yandex في قائمة وحدة التحكم.

أين تجد الزر

انقر فوق الزر . هنا يمكننا تعيين اسم القاعدة ونوعها. كنوع ، أوصي باختيار 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, , .

في الخطوة التالية ، أخطط لتطوير قائمة وتنفيذ تطبيق بالفعل حيث سيكون من الممكن ببساطة وضع أي أوامر.