[AWS] Lambda programming for DynamoDB

Create Table

import boto3
 
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
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
    }
)
 
print("Table status:", table.table_status)
 

Insert Data

import boto3
import json
import decimal
 
# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, decimal.Decimal):
            if abs(o) % 1 > 0:
                return float(o)
            else:
                return int(o)
        return super(DecimalEncoder, self).default(o)
 
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
 
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)
        }
    }
)
 
print("PutItem succeeded:")
print(json.dumps(response, indent=4, cls=DecimalEncoder))
 

Select Data

from __future__ import print_function # Python 2/3 compatibility
import boto3
import json
import decimal
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):
            return str(o)
        return super(DecimalEncoder, self).default(o)
 
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
 
table = dynamodb.Table('Movies')
 
print("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')
)
 
for i in response[u'Items']:
    print(json.dumps(i, cls=DecimalEncoder))
 

Update Data

import boto3
import json
import decimal
 
# 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)
 
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
 
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"
)
 
print("UpdateItem succeeded:")
print(json.dumps(response, indent=4, cls=DecimalEncoder))
 

[출처] https://www.qwiklabs.com