DynamoDB Test with Python

dynamodb_test.py

import boto3

dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-2')

# Create table
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)

# Put Item

import decimal

table = dynamodb.Table('Movies')

table.put_item(
    Item={
        "year" : 2013,
        "title" : "Turn It Down, Or Else!",
        "info" : {
            "directors" : [
                "Alice Smith",
                "Bob Jones"
            ],
            "release_date" : "2013-01-18T00:00:00Z",
            "rating" : decimal.Decimal('6.2'),
            "genres" : [
                "Comedy",
                "Drama"
            ],
            "image_url" : "http://ia.media-imdb.com/images/N/O9ERWAU7FS797AJ7LU8HN09AMUP908RLlo5JF90EWR7LJKQ7@@._V1_SX400_.jpg",
            "plot" : "A rock band plays their music at high volumes, annoying the neighbors.",
            "rank" : 11,
            "running_time_secs" : 5215,
            "actors" : [
                "David Matthewman",
                "Ann Thomas",
                "Jonathan G. Neff"
            ]
        }
    }
)

# Get Item
# Primary Key로 레코드를 찾아 알려줌 (Partition Key 필수, Sort Key 있으면 함께 적어야 함)
import json

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)

response = table.get_item(
    Key={
        'year': 2013,
        'title': "Turn It Down, Or Else!"
    }
)

print(json.dumps(response['Item'], cls=DecimalEncoder))

# Update item
# (https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.update_item)
# (https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.update_item)
# AttributeUpdates는 Legacy Parameter이므로, UpdateExpression 속성을 이용하는 것을 권장함
# UpdateExpression 시 구문은 https://docs.aws.amazon.com/ko_kr/amazondynamodb/latest/developerguide/Expressions.UpdateExpressions.html 참고

response = table.update_item(
    Key={
        'year': 2013,
        'title': "Turn It Down, Or Else!"
    },
    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, cls=DecimalEncoder))

# 카운터 증가(또는 감소) - 숫자인 경우
# - UpdateExpression="set info.rating = info.rating + :val"
# - ExpressionAttributeValues={ ':val': decimal.Decimal(1)}

# 조건부 업데이트 (List의 길이가 3보다 큰 경우)
# - UpdateExpression="remove info.actors[0]"
# - ConditionExpression="size(info.actors) > :num"
# - ExpressionAttributeValues={ ':num': 3 }

# Delete item
# 기본 키를 지정하여 항목 한 개 삭제 (ConditionExpression 적용 가능)
response = table.delete_item(
    Key={
        'year': 2013,
        'title': "Turn It Down, Or Else!"
    },
    ConditionExpression="info.rating <= :val",
    ExpressionAttributeValues={
        ":val": decimal.Decimal(5)
    }
)

# (참고: 조건에 안 맞으므로 실행 안됨)

# 쿼리 하기 1: 조건에 맞는 것만
from boto3.dynamodb.conditions import Key, Attr

response = table.query(
    KeyConditionExpression=Key('year').eq(2013)
)

for i in response['Items']:
    print(i['year'], ':', i['title'])

# 쿼리 하기 2: 특정 년도 & 제목이 특정 값 사이에 있는 것들만
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(2013) & Key('title').between('A', 'U')
)

for i in response['Items']:
    print(json.dumps(i, cls=DecimalEncoder))

# Scan: 테이블 전체의 항목 읽고 모든 데이터 반환 
# (선택 사항으로 filter_expression 제공 가능하며 일치하는 것만 반환)

response = table.scan(
    FilterExpression=Key('year').between(1950, 1959),
    ProjectionExpression="#yr, title, info.rating",
    ExpressionAttributeNames={ "#yr": "year" }
)

# scan 메서드는 언제나 항목의 하위 집합(page)을 반환함. 
# 마지막 값(LastEvaluatedKey)이 scan 파라미터를 통해 scan 파라미터를 통해 ExclusiveStartKey 메서드에 전달

for i in response['Items']:
    print(json.dumps(i, cls=DecimalEncoder))

while 'LastEvaluatedKey' in response:
    response = table.scan(
        FilterExpression=Key('year').between(1950, 1959),
        ProjectionExpression="#yr, title, info.rating",
        ExpressionAttributeNames={ "#yr": "year" },
        ExclusiveStartKey=response['LastEvaluatedKey']
        )

    for i in response['Items']:
        print(json.dumps(i, cls=DecimalEncoder))

# 테이블 삭제

table.delete()