# -*- coding:utf-8 -*-
# @Time : 3/23/2020 11:42 AM
# @Author : Mandy Lin
# @File : call_dynamo.py
# @Software : PyCharm
import boto3
from pprint import pprint
def __reform_item(item):
new_item = {}
for c in item.keys():
if isinstance(item[c], str):
new_item[c] = {'S': item[c]}
if isinstance(item[c], int):
new_item[c] = {'N': str(item[c])}
return new_item
# use boto3 directly
session = boto3.Session(profile_name='default', region_name='us-east-1')
my_db = session.client(service_name='dynamodb')
def scan_item():
table = my_db.scan(TableName='Music',
FilterExpression='Artist = :a',
ExpressionAttributeValues={
':a': {
'S': 'Acme Band 2'
},
},)
pprint(table)
print('----------------------------------------')
# scan_item()
Item = {'ObjectId': '7370e2ae-6da3-11ea-b2d6-989096dd40fd',
'Price': 399,
'ItemDescription': 'mouse'}
def write_item(item):
item = __reform_item(item)
resp = my_db.put_item(TableName='demo_db', Item=item)
print('--------input item----------------')
print(item)
print('--------output resp----------------')
print(resp)
# write_item(Item)
key = {'ObjectId': 'c8e537026da011eabf53989096dd40fd'}
# key = {'ItemDescription': 'mouse'}
# key should be the primary key
def get_item(key):
key = __reform_item(key)
resp = my_db.get_item(TableName='demo_db', Key=key)
pprint(resp)
print('---------------------------')
# get_item(key)
def tag_resource(ResourceArn, Tags):
resp = my_db.tag_resource(ResourceArn=ResourceArn, Tags=Tags)
pprint(resp)
# user should be authorized to perform
# tag_resource(ResourceArn='demo_db', Tags=[{'Key': 'demoKey', 'Value': 'demoValue'}])
def update_item_column(Key, UpdateExpression, ExpressionAttributeValues):
key_ = __reform_item(Key)
resp = my_db.update_item(TableName='demo_db', Key=key_, UpdateExpression=UpdateExpression, ExpressionAttributeValues=ExpressionAttributeValues, ReturnValues='UPDATED_NEW')
pprint(resp)
Key = {'ObjectId': '7370e2ae-6da3-11ea-b2d6-989096dd40fd'}
UpdateExpression = 'SET ItemDescription = :label'
# UpdateExpression = 'REMOVE ItemDescription'
ExpressionAttributeValues = {':label': {'S': 'mouse22220--00'}}
# ExpressionAttributeNames =
# update item's value for a specific column
# update_item_column(Key, UpdateExpression, ExpressionAttributeValues)
def create_table():
response = my_db.create_table(AttributeDefinitions=[
{
'AttributeName': 'Artist',
'AttributeType': 'S',
},
{
'AttributeName': 'SongTitle',
'AttributeType': 'S',
}
],
KeySchema=[
{
'AttributeName': 'Artist',
'KeyType': 'HASH',
},
{
'AttributeName': 'SongTitle',
'KeyType': 'RANGE',
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
},
TableName='Music'
)
pprint(response)
print('-------------------')
# create_table()
def batch_add_item():
response = my_db.batch_write_item(
RequestItems={
'Music': [
{
'PutRequest': {
'Item': {
'AlbumTitle': {
'S': 'No One You Know 4',
},
'Artist': {
'S': 'Python',
},
'SongTitle': {
'S': 'hello py world',
},
},
},
},
{
'PutRequest': {
'Item': {
'AlbumTitle': {
'S': 'Move theme',
},
'Artist': {
'S': 'Jennifer Lawrence',
},
'SongTitle': {
'S': 'Hunger Game',
},
},
},
},
{
'PutRequest': {
'Item': {
'AlbumTitle': {
'S': 'Blue Planet',
},
'Artist': {
'S': 'David Attenborough',
},
'SongTitle': {
'S': 'David Attenborough',
},
},
},
},
],
},
)
pprint(response)
# pprint(my_db.describe_table(TableName='Music'))
batch_add_item()
def batch_delete_item():
response = my_db.batch_write_item(
RequestItems={
'Music': [
{
'DeleteRequest': {
'Key': {
'Artist': {
'S': 'Acme Band',
},
'SongTitle': {
'S': 'Happy Day',
},
},
},
},
],
},
)
pprint(response)
def batch_get_item():
response = my_db.batch_get_item(
RequestItems={
'Music': {
'Keys': [
{
'Artist': {
'S': 'No One You Know',
},
'SongTitle': {
'S': 'Call Me Today',
}
},
],
'AttributesToGet': [
'AlbumTitle',
],
'ConsistentRead': True,
}
}
)
pprint(response)
# batch_delete_item()
# batch_get_item()
