介绍性
这篇文章的延续这篇文章。在其中,我们检查了电报bot的yandex云功能的创建和配置。今天,我们将考虑将机器人的电报连接到数据库,并保存有关与机器人通信的用户的任何信息。
我们将使用Yandex Cloud Database作为数据库。
任务
创建一个数据库;
准备连接底座;
安装依赖项;
向数据库添加表以存储用户;
将有关传入用户的信息保存在电报消息中;
获取信息并将其从数据库发送给用户。
数据库创建
我们列表中最简单的任务,我们需要转到我们帐户下的Yandex Cloud Console。然后在控制台菜单中选择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 .
, . .
下一步,我计划开发一个菜单,并且已经实现了一个应用程序,可以在其中轻松下订单。