Let's try various services recommended by AWS serverless. This time, let's use Amazon DynamoDB from Python.
There is no environment in the first place !! So, it is from the environment construction.
Build the environment by referring to the following (or as it is ...).
If you do not set .bash_profile, it seems that you have to hit the command to switch the Python version every time, so set this as well.
Create a Python script by referring to the following.
Since I'm from a local Mac, IAM Role isn't available, and I wanted to complete it all with a single script, so I'll use this authentication method.
python
import boto3
from boto3.session import Session
accesskey = "YOUR_ACCESSKEY"
secretkey = "YOUR_SECRETKEY"
region = "YOUR_REGION"
session = Session(
aws_access_key_id=accesskey,
aws_secret_access_key=secretkey,
region_name=region)
Confirm the operation as AWS says.
Below is a sample. I confirmed it by commenting on the call to the main function.
To be honest, I'm not used to Python, so I'm not sure if this is the way to write it ...
SamplaDynamoDB.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----
# ==============================================================================
#
# SampleDynamoDB.py
#
# * ex) $ ./SamplaDynamoDB.py
#
# ==============================================================================
import sys
import io
import logging
import json
import decimal
import time
import datetime
from datetime import datetime as dt
import boto3
from boto3.session import Session
from boto3.dynamodb.conditions import Key, Attr
# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, decimal.Decimal):
if o % 1 > 0:
return float(o)
else:
return int(o)
return super(DecimalEncoder, self).default(o)
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
logging.basicConfig(level=logging.INFO, filename=(__file__ + ".log"), format="%(asctime)s %(levelname)s %(filename)s %(lineno)d %(funcName)s | %(message)s")
# ------------------------------------------------------------------------------
# Set
# ------------------------------------------------------------------------------
tmp_today = datetime.datetime.today()
# AWS
accesskey = "[Access key ID]"
secretkey = "[Secret access key]"
region = "ap-northeast-1"
session = Session(aws_access_key_id=accesskey, aws_secret_access_key=secretkey, region_name=region)
dynamodb = session.resource('dynamodb')
# ------------------------------------------------------------------------------
#"step 1:Create a table- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.01.html>
# ------------------------------------------------------------------------------
def MoviesCreateTable():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.create_table(
TableName='Movies',
KeySchema=[
{
'AttributeName': 'year',
'KeyType': 'HASH' #Partition key
},
{
'AttributeName': 'title',
'KeyType': 'RANGE' #Sort key
}
],
AttributeDefinitions=[
{
'AttributeName': 'year',
'AttributeType': 'N'
},
{
'AttributeName': 'title',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
)
logging.info("Table status : [%s]", table.table_status)
except Exception as e:
logging.error("Type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ------------------------------------------------------------------------------
#"Step 2:Load sample data- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.02.html>
# ------------------------------------------------------------------------------
def MoviesLoadData():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
with open("moviedata.json") as json_file:
movies = json.load(json_file, parse_float = decimal.Decimal)
for movie in movies:
year = int(movie['year'])
title = movie['title']
info = movie['info']
#logging.info("Adding Movie | Year:[%s], Title:[%s]", year, title)
table.put_item(
Item={
'year': year,
'title': title,
'info': info,
}
)
except Exception as e:
logging.error("Type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ------------------------------------------------------------------------------
#"Step 3:Create, load, update, delete items- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.03.html>
# ------------------------------------------------------------------------------
def MoviesItemOps01():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
response = table.put_item(
Item={
'year': year,
'title': title,
'info': {
'plot':"Nothing happens at all.",
'rating': decimal.Decimal(0)
}
}
)
logging.info("PutItem succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesItemOps02():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
try:
response = table.get_item(
Key={
'year': year,
'title': title
}
)
except ClientError as e:
logging.info(e.response['Error']['Message'])
else:
item = response['Item']
logging.info("GetItem succeeded:")
logging.info(json.dumps(item, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesItemOps03():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
response = table.update_item(
Key={
'year': year,
'title': title
},
UpdateExpression="set info.rating = :r, info.plot=:p, info.actors=:a",
ExpressionAttributeValues={
':r': decimal.Decimal(5.5),
':p': "Everything happens all at once.",
':a': ["Larry", "Moe", "Curly"]
},
ReturnValues="UPDATED_NEW"
)
logging.info("UpdateItem succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesItemOps04():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
response = table.update_item(
Key={
'year': year,
'title': title
},
UpdateExpression="set info.rating = info.rating + :val",
ExpressionAttributeValues={
':val': decimal.Decimal(1)
},
ReturnValues="UPDATED_NEW"
)
logging.info("UpdateItem succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesItemOps05():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
logging.info("Attempting conditional update...")
try:
response = table.update_item(
Key={
'year': year,
'title': title
},
UpdateExpression="remove info.actors[0]",
ConditionExpression="size(info.actors) > :num",
ExpressionAttributeValues={
':num': 2
},
ReturnValues="UPDATED_NEW"
)
except ClientError as e:
if e.response['Error']['Code'] == "ConditionalCheckFailedException":
logging.error(e.response['Error']['Message'])
else:
raise
else:
logging.info("UpdateItem succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesItemOps06():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
title = "The Big New Movie"
year = 2015
logging.info("Attempting a conditional delete...")
try:
response = table.delete_item(
Key={
'year': year,
'title': title
},
ConditionExpression="info.rating <= :val",
ExpressionAttributeValues= {
":val": decimal.Decimal(8)
}
)
except ClientError as e:
if e.response['Error']['Code'] == "ConditionalCheckFailedException":
logging.info(e.response['Error']['Message'])
else:
raise
else:
logging.info("DeleteItem succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
except Exception as e:
logging.error("type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ------------------------------------------------------------------------------
#"Step 4:Query and scan data- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.04.html>
# ------------------------------------------------------------------------------
def MoviesQuery01():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
logging.info("Movies from 1933")
response = table.query(
KeyConditionExpression=Key('year').eq(1933)
)
logging.info("Query01 succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
for i in response['Items']:
logging.info("%s : %s", i['year'], i['title'])
except Exception as e:
logging.error("Type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
def MoviesQuery02():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
logging.info("Movies from 1992 - titles A-L, with genres and lead actor")
response = table.query(
ProjectionExpression="#yr, title, info.genres, info.actors[0]",
ExpressionAttributeNames={ "#yr": "year" }, # Expression Attribute Names for Projection Expression only.
KeyConditionExpression=Key('year').eq(1992) & Key('title').between('A', 'L')
)
logging.info("Query02 succeeded:")
logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
for i in response[u'Items']:
logging.info(json.dumps(i, cls=DecimalEncoder))
except Exception as e:
logging.error("Type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ------------------------------------------------------------------------------
#"Step 5: (option)Drop the table- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.05.html>
# ------------------------------------------------------------------------------
def MoviesDeleteTable():
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
try:
table = dynamodb.Table('Movies')
table.delete()
except Exception as e:
logging.error("Type : %s", type(e))
logging.error(e)
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ------------------------------------------------------------------------------
# Main
# ------------------------------------------------------------------------------
if __name__ == '__main__':
logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
# Set
logging.info("REGION : [%s]", region)
# Args
logging.info("Argc : [%d]", len(sys.argv))
for i in range(len(sys.argv)):
logging.info("Argv[%d] : [%s]", i, sys.argv[i])
# Check Table
table = dynamodb.Table('Movies')
logging.info("Table :")
logging.info(table)
# Create Table
#MoviesCreateTable()
#time.sleep(9)
# Load Json Data
#MoviesLoadData()
# Query Data
MoviesItemOps01()
MoviesItemOps02()
MoviesItemOps03()
MoviesItemOps04()
#MoviesItemOps05()
#MoviesItemOps06()
# Query Data
MoviesQuery01()
MoviesQuery02()
# Delete Table
#MoviesDeleteTable()
logging.info("<<<<<<<< %s End >>>>>>>>", __name__)
# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----
This time, I was grateful that all the sources were in the introductory guide.
Next, I thought I'd try launching Amazon Lambda.
Recommended Posts