ПР. Запуск тестового приложения
Кратко:
- Запуск тестового приложения с использованием AWS SDK для Python
- Создание таблиц БД и запись и чтение данных
- Создание таблиц с использованием AWS SDK для Python
- Вставка данных в таблицу с использованием AWS SDK для Python
- Удаление таблицы с использованием AWS SDK для Python
Практическая работа. Запуск тестового приложения
В предыдущем уроке вы прошли подготовительные этапы: создали и настроили сервисный аккаунт, выпустили статический ключ, а также научились работать с таблицами и данными с помощью низкоуровневого API и CLI.
В этом уроке вы продолжите работу с инструментами AWS и с помощью AWS SDK для языка Python научитесь выполнять такие базовые операции, как создание таблиц БД, запись и чтение данных.
Для выполнения работы вам понадобится Python версии 3.6 и выше и библиотека boto3.
Установить эту библиотеку можно с помощью команды:
pip install boto3
Создание таблицы
Создайте файл с именем
SeriesCreateTable.py
и скопируйте в него исходный код программы:
import boto3
def create_series_table():
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.create_table(
TableName = 'docapitest/series', # Series — имя таблицы
KeySchema = [
{
'AttributeName': 'series_id',
'KeyType': 'HASH' # Ключ партицирования
},
{
'AttributeName': 'title',
'KeyType': 'RANGE' # Ключ сортировки
}
],
AttributeDefinitions = [
{
'AttributeName': 'series_id',
'AttributeType': 'N' # Целое число
},
{
'AttributeName': 'title',
'AttributeType': 'S' # Строка
},
]
)
return table
if __name__ == '__main__':
series_table = create_series_table()
print("Table status:", series_table.table_status)
С помощью консоли управления убедитесь, что в директории
docapitest
появилась таблица series
.Первоначальная загрузка данных
Для того чтобы вставить данные в созданную таблицу
series
, создайте файл с именем SeriesLoadData.py
и скопируйте в него следующий исходный код программы:
from decimal import Decimal
import json
import boto3
def load_series(series):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document API эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
for serie in series:
series_id = int(serie['series_id'])
title = serie['title']
print("Series added:", series_id, title)
table.put_item(Item = serie)
if __name__ == '__main__':
with open("seriesdata.json") as json_file:
serie_list = json.load(json_file, parse_float = Decimal)
load_series(serie_list)
Отредактируйте файл
SeriesLoadData.py
и укажите значение endpoint_url
вашей базы.Для загрузки данных приложение будет использовать данные, которые записаны в файл
seriesdata.json
. Создайте этот файл и скопируйте в него описание сериалов:
[{
"series_id": 1,
"title": "IT Crowd",
"info": {
"release_date": "2006-02-03T00:00:00Z",
"series_info": "The IT Crowd is a British sitcom produced by Channel 4, written by Graham Linehan, produced by Ash Atalla and starring Chris O'Dowd, Richard Ayoade, Katherine Parkinson, and Matt Berry"
}
},
{
"series_id": 2,
"title": "Silicon Valley",
"info": {
"release_date": "2014-04-06T00:00:00Z",
"series_info": "Silicon Valley is an American comedy television series created by Mike Judge, John Altschuler and Dave Krinsky. The series focuses on five young men who founded a startup company in Silicon Valley"
}
},
{
"series_id": 3,
"title": "House of Cards",
"info": {
"release_date": "2013-02-01T00:00:00Z",
"series_info": "House of Cards is an American political thriller streaming television series created by Beau Willimon. It is an adaptation of the 1990 BBC miniseries of the same name and based on the 1989 novel of the same name by Michael Dobbs"
}
},
{
"series_id": 3,
"title": "The Office",
"info": {
"release_date": "2005-03-24T00:00:00Z",
"series_info": "The Office is an American mockumentary sitcom television series that depicts the everyday work lives of office employees in the Scranton, Pennsylvania, branch of the fictional Dunder Mifflin Paper Company"
}
},
{
"series_id": 3,
"title": "True Detective",
"info": {
"release_date": "2014-01-12T00:00:00Z",
"series_info": "True Detective is an American anthology crime drama television series created and written by Nic Pizzolatto. The series, broadcast by the premium cable network HBO in the United States, premiered on January 12, 2014"
}
},
{
"series_id": 4,
"title": "The Big Bang Theory",
"info": {
"release_date": "2007-09-24T00:00:00Z",
"series_info": "The Big Bang Theory is an American television sitcom created by Chuck Lorre and Bill Prady, both of whom served as executive producers on the series, along with Steven Molaro"
}
},
{
"series_id": 5,
"title": "Twin Peaks",
"info": {
"release_date": "1990-04-08T00:00:00Z",
"series_info": "Twin Peaks is an American mystery horror drama television series created by Mark Frost and David Lynch that premiered on April 8, 1990, on ABC until its cancellation after its second season in 1991 before returning as a limited series in 2017 on Showtime"
}
}
]
Запустите программу (для её успешного выполнения может понадобиться указать в скрипте
SeriesLoadData.py
полный путь к файлу seriesdata.json
):
python SeriesLoadData.py
В результате выполнения вы увидите вывод программы:
Series added: 1 IT Crowd
Series added: 2 Silicon Valley
Series added: 3 House of Cards
Series added: 3 The Office
Series added: 3 True Detective
Series added: 4 The Big Bang Theory
Series added: 5 Twin Peaks
Работа с записями
Создание записи
Теперь создайте файл
SeriesItemPut.py
и скопируйте в него следующий код:
from pprint import pprint
import boto3
def put_serie(series_id, title, release_date, series_info):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
response = table.put_item(
Item = {
'series_id': series_id,
'title': title,
'info': {
'release_date': release_date,
'series_info': series_info
}
}
)
return response
if __name__ == '__main__':
serie_resp = put_serie(3, "Supernatural", "2015-09-13",
"Supernatural is an American television series created by Eric Kripke")
print("Series added successfully:")
pprint(serie_resp, sort_dicts = False)
В результате выполнения этого кода в таблице добавится запись о сериале Supernatural.
Чтение записи
Создайте файл
SeriesItemGet.py
и скопируйте в него следующий код:
from pprint import pprint
import boto3
from botocore.exceptions import ClientError
def get_serie(title, series_id):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
try:
response = table.get_item(Key = {'series_id': series_id, 'title': title})
except ClientError as e:
print(e.response['Error']['Message'])
else:
return response['Item']
if __name__ == '__main__':
serie = get_serie("Supernatural", 3,)
if serie:
print("Record read:")
pprint(serie, sort_dicts = False)
Результатом будет сообщение в формате JSON с данными о сериале.
Обновление записи
В файле
SeriesItemUpdate.py
разместите код обновления записи:
from decimal import Decimal
from pprint import pprint
import boto3
def update_serie(title, series_id, release_date, rating):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
response = table.update_item(
Key = {
'series_id': series_id,
'title': title
},
UpdateExpression = "set info.release_date = :d, info.rating = :r ",
ExpressionAttributeValues = {
':d': release_date,
':r': Decimal(rating)
},
ReturnValues = "UPDATED_NEW"
)
return response
if __name__ == '__main__':
update_response = update_serie(
"Supernatural", 3, "2005-09-13", 8)
print("Series updated:")
pprint(update_response, sort_dicts = False)
Результатом будет сообщение в формате JSON с измененными данными.
Удаление записи
Создайте файл
SeriesItemDelete.py
и скопируйте в него следующий код:
from decimal import Decimal
from pprint import pprint
import boto3
from botocore.exceptions import ClientError
def delete_underrated_serie(title, series_id, rating):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
try:
response = table.delete_item(
Key = {
'series_id': series_id,
'title': title
},
ConditionExpression = "info.rating <= :val",
ExpressionAttributeValues = {
":val": Decimal(rating)
}
)
except ClientError as e:
if e.response['Error']['Code'] == "ConditionalCheckFailedException":
print(e.response['Error']['Message'])
else:
raise
else:
return response
if __name__ == '__main__':
print("Deleting...")
delete_response = delete_underrated_serie("Supernatural", 3, 8)
if delete_response:
print("Series data deleted:")
pprint(delete_response, sort_dicts = False)
Убедитесь, что данные о сериале Supernatural удалены из таблицы.
Поиск по ключам партицирования и сортировки
Код поиска разместите в новом файле
SeriesQuery.py
:
from pprint import pprint
import boto3
from boto3.dynamodb.conditions import Key
def query_and_project_series(series_id, title_range):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
response = table.query(
ProjectionExpression = "series_id, title, info.release_date",
KeyConditionExpression = Key('series_id').eq(series_id) & Key('title').begins_with(title_range)
)
return response['Items']
if __name__ == '__main__':
query_id = 3
query_range = 'T'
print(f"Series with ID = {query_id} and names beginning with "
f"{query_range}")
series = query_and_project_series(query_id, query_range)
for serie in series:
print(f"\n{serie['series_id']} : {serie['title']}")
pprint(serie['info'])
Результатом будет сообщение:
Series with ID = 3 and names beginning with T
3 : The Office
{'release_date': '2005-03-24T00:00:00Z'}
3 : True Detective
{'release_date': '2014-01-12T00:00:00Z'}
Запуск операции Scan
Создайте файл
SeriesTableScan.py
и скопируйте в него следующий код:
from pprint import pprint
import boto3
from boto3.dynamodb.conditions import Key
def scan_series(id_range, display_series):
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
scan_kwargs = {
'FilterExpression': Key('series_id').between(*id_range),
'ProjectionExpression': "series_id, title, info.release_date"
}
done = False
start_key = None
while not done:
if start_key:
scan_kwargs['ExclusiveStartKey'] = start_key
response = table.scan(**scan_kwargs)
display_series(response.get('Items', []))
start_key = response.get('LastEvaluatedKey', None)
done = start_key is None
if __name__ == '__main__':
def print_series(series):
for serie in series:
print(f"\n{serie['series_id']} : {serie['title']}")
pprint(serie['info'])
query_range = (1, 3)
print(f"Series with IDs from {query_range[0]} to {query_range[1]}...")
scan_series(query_range, print_series)
Результатом будет сообщение:
Series with IDs from 1 to 3...
3 : House of Cards
{'release_date': '2013-02-01T00:00:00Z'}
3 : The Office
{'release_date': '2005-03-24T00:00:00Z'}
3 : True Detective
{'release_date': '2014-01-12T00:00:00Z'}
1 : IT Crowd
{'release_date': '2006-02-03T00:00:00Z'}
2 : Silicon Valley
{'release_date': '2014-04-06T00:00:00Z'}
Удаление таблицы
Создайте файл
SeriesTableDelete.py
и скопируйте в него следующий код:
import boto3
def delete_serie_table():
ydb_docapi_client = boto3.resource('dynamodb', endpoint_url = "<Document_API_эндпоинт>")
table = ydb_docapi_client.Table('docapitest/series')
table.delete()
if __name__ == '__main__':
delete_serie_table()
print("Table Series deleted")
Убедитесь, что таблица удалена из базы данных.
Поздравляем, вы завершили изучение темы Serverless YDB!