LoginSignup
18
22

More than 5 years have passed since last update.

@kusokamayarou

Amazon DynamoDB を Python から使ってみる

Posted at

０．はじめに

AWS がサーバーレスで推奨しているサービスを、色々と使ってみます。
今回は、Amazon DynamoDB を Python から使ってみます。

１．Mac に Python 環境を構築する

そもそも環境が無い!! ってことで、環境構築からです。

以下を参考にして(っていうか、そのままですが…)、環境を構築します。

.bash_profile の設定をしておかないと、Python のバージョンの切替のコマンドを毎回叩かないといけないみたいなので、こちらも設定します。

２．Mac の Python 環境から、Amazon DynamoDB を叩くスクリプトを作成する

以下を参考にして、Python スクリプトを作成します。

ローカルの Mac からですので、IAM Role は利用出来ないですし、スクリプト一つで全て完結させたかったので、こちらの認証のやり方を使います。

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)

３．AWS Documentation の Amazon DynamoDB 入門ガイド に沿って、DynamoDB への操作を確認する

もう AWS さんの仰る通りに操作を確認します。

  1. ステップ 1: テーブルを作成する - Amazon DynamoDB
  2. ステップ 2: サンプルデータをロードする - Amazon DynamoDB
  3. ステップ 3: 項目を作成、読み込み、更新、削除する - Amazon DynamoDB
  4. ステップ 4: データをクエリおよびスキャンする - Amazon DynamoDB
  5. ステップ 5: (オプション) テーブルを削除する - Amazon DynamoDB

以下、サンプルです。
メイン関数の呼び出しをコメントにしたりして、確認しました。

正直、Python 使い慣れていないので、こういう書き方でいいのかよくわかりませんが…。

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 = "[アクセスキー ID]"
secretkey = "[シークレットアクセスキー]"
region = "ap-northeast-1"
session = Session(aws_access_key_id=accesskey, aws_secret_access_key=secretkey, region_name=region)
dynamodb = session.resource('dynamodb')

# ------------------------------------------------------------------------------
# 「ステップ 1: テーブルを作成する - 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__)

# ------------------------------------------------------------------------------
# 「ステップ 2: サンプルデータをロードする - 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__)

# ------------------------------------------------------------------------------
# 「ステップ 3: 項目を作成、読み込み、更新、削除する - 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__)

# ------------------------------------------------------------------------------
# 「ステップ 4: データをクエリおよびスキャンする - 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__)

# ------------------------------------------------------------------------------
# 「ステップ 5: (オプション) テーブルを削除する - 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----

９９．ハマりポイント

  • 入門ガイドのソースをそのまま貼っ付けて確認しただけなので、ほぼ無いです。
  • ただ、ちゃんと書こうと思ってロギングの機能を盛り込んだんですが、何せ Python あまり使ったことが無いので、その辺引っ掛かったりしました。

ＸＸ．まとめ

今回は、とにかく全部入門ガイドにソースがあったので、有難かったです。

次は、Amazon Lambda 起動を試そうかなと。

18
22
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
18
22