AWS SDK boto3 / DynamoDB [Python3]

Boto3 is the Amazon Web Services (AWS) Software Development Kit (SDK) for Python.
In this article, I would like to share how to access DynamoDB by Boto3/Python3.

😎 Installation

pip install boto3

😀 Get DynamoDB Resource object

import boto3

@lru_cache(maxsize=1)
def get_dynamodb_resource():
    rds = boto3.client('dynamodb', region_name='us-west-2')

🍮 Create table if there is no table

def create_table():
    dynamodb = get_dynamodb_resource()

    try:
        table = dynamodb.create_table(
            TableName='article',
            KeySchema=[
                 {'AttributeName': 'id', 'KeyType': 'HASH'},
            ],
            AttributeDefinitions=[
                {'AttributeName': 'id', 'AttributeType': 'N'},
            ],
            ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
        )
        # Wait until the table exists.
        table.meta.client.get_waiter('table_exists').wait(TableName=table_name)
        print('created table {}.'.format(table_name))
    except ClientError as e:
        # If there is the table already, the error will be skipped
        if e.response['Error']['Code'] != 'ResourceInUseException':
            raise e

🐡 Put Item

def save_articles(id):
    dynamodb = get_dynamodb_resource()
    table = dynamodb.Table('articles')

    table.put_item(
        Item={
            'id': id,
            'title': 'sample-title',
            'content': 'sample-content',
            'created_at': arrange_datetime(round(time.time() * 1000)),
        }
    )

😼 Batch Writing

def save_all_articles(items):
    dynamodb = get_dynamodb_resource()
    table = dynamodb.Table(D_JOURNAL_TABLE)

    with table.batch_writer() as batch:
        for item in items:
            arranged_item = {
                'id': item['id'],
                'title': item['title'],
                'content': item['content'],
                'created_at': arrange_datetime(round(time.time() * 1000)),
            }
            batch.put_item(Item=arranged_item)

🤔 Get Item

def get_article(id):
    dynamodb = get_dynamodb_resource()

    table = dynamodb.Table('articles')
    response = table.get_item(
        Key={
            'id': str(id),
        }
    )
    if 'Item' in response:
      return response['Item']
    else:
      return None

🎃 References

• https://boto3.readthedocs.io/en/latest/guide/dynamodb.html
• http://blog.brains-tech.co.jp/entry/2015/09/30/222148
• https://qiita.com/estaro/items/b329deafdfef790aa355
• https://qiita.com/kusokamayarou/items/fc34447eb6970da1a103
• https://qiita.com/hshimo/items/e5ad98b21786d796f1da

🖥 Recommended VPS Service

VULTR provides high performance cloud compute environment for you. Vultr has 15 data-centers strategically placed around the globe, you can use a VPS with 512 MB memory for just $ 2.5 / month ($ 0.004 / hour). In addition, Vultr is up to 4 times faster than the competition, so please check it => Check Benchmark Results!!