Dynamodbtest

wordpress meta

title: 'DynamoDB Test'
date: '2017-10-14T10:23:20-05:00'
status: publish
permalink: /dynamodbtest
author: admin
excerpt: ''
type: post
id: 1138
category:
    - DynamoDB
    - Python
tag: []
post_format: []

Boto3 and AWS DynamoDB usage...

http://boto3.readthedocs.io/en/latest/reference/services/dynamodb.html

$ cat dynamodbTest.py 
import boto3

#dynamodb = boto3.resource('dynamodb')
# Hard coding strings as credentials, not recommended. Use configs or env variables AWS_ACCESS_KEY, AWS_SECRET_KEY
dynamodb = boto3.resource(
    'dynamodb',
    aws_access_key_id='KEY_ID_REMOVED',
    aws_secret_access_key='ACCESS_KEY_REMOVED',
    region_name = 'us-east-1'
)

def create_table(tableName):
  table = dynamodb.create_table(
    TableName=tableName,
    KeySchema=[
        {
            'AttributeName': 'username', 
            'KeyType': 'HASH'
        },
        {
            'AttributeName': 'last_name', 
            'KeyType': 'RANGE'
        }
    ], 
    AttributeDefinitions=[
        {
            'AttributeName': 'username', 
            'AttributeType': 'S'
        }, 
        {
            'AttributeName': 'last_name', 
            'AttributeType': 'S'
        }, 
    ], 
    ProvisionedThroughput={
        'ReadCapacityUnits': 1, 
        'WriteCapacityUnits': 1
    }
  )

  table.meta.client.get_waiter('table_exists').wait(TableName=tableName)
  print 'Table item count: {}'.format(table.item_count)

def delete_table(tableName):
  table = dynamodb.Table(tableName)
  table.delete()

def put_item(tableName):
  table = dynamodb.Table(tableName)

  response = table.put_item(
   Item={
        'username': 'jdoe',
        'first_name': 'jane',
        'last_name': 'doe',
        'age': 20,
        'account_type': 'librarian',
    }
  )

  print response

def get_item(tableName):
  table = dynamodb.Table(tableName)

  response = table.get_item(
   Key={
        'username': 'jdoe',
        'last_name': 'doe'
    }
  )

  item = response['Item']
  name = item['first_name']

  print(item)
  print("Hello, {}" .format(name))

def update_item(tableName):
  table = dynamodb.Table(tableName)

  table.update_item(
    Key={
        'username': 'jdoe',
        'last_name': 'doe'
    },
    UpdateExpression='SET age = :val1',
    ExpressionAttributeValues={
        ':val1': 23
    }
  )

def delete_item(tableName):
  table = dynamodb.Table(tableName)

  table.delete_item(
    Key={
        'username': 'jdoe',
        'last_name': 'doe'
    }
  )

def batch_write(tableName):
  table = dynamodb.Table(tableName)

  with table.batch_writer() as batch:
    batch.put_item(
        Item={
            'account_type': 'end_user',
            'username': 'bbob',
            'first_name': 'billy',
            'last_name': 'bob',
            'age': 20,
            'address': {
                'road': '1 fake street',
                'city': 'Houston',
                'state': 'TX',
                'country': 'USA'
            }
        }
    )
    batch.put_item(
        Item={
            'account_type': 'librarian',
            'username': 'user1',
            'first_name': 'user1 first name',
            'last_name': 'user1 last name',
            'age': 20,
            'address': {
                'road': '10 fake street',
                'city': 'Dallas',
                'state': 'TX',
                'country': 'USA'
            }
        }
    )
    batch.put_item(
        Item={
            'account_type': 'end_user',
            'username': 'user2',
            'first_name': 'user2 first name',
            'last_name': 'user2 last name',
            'age': 23,
            'address': {
                'road': '12 fake street',
                'city': 'Austin',
                'province': 'TX',
                'state': 'USA'
            }
        }
    )

def create_multiple_items(tableName,itemCount):

  table = dynamodb.Table(tableName)

  with table.batch_writer() as batch:
    for i in range(itemCount):
        batch.put_item(
            Item={
                'account_type': 'anonymous',
                'username': 'user-' + str(i),
                'first_name': 'unknown',
                'last_name': 'unknown'
            }
        )


def query(tableName):
  from boto3.dynamodb.conditions import Key, Attr
  table = dynamodb.Table(tableName)

  response = table.query(
    KeyConditionExpression=Key('username').eq('user2')
  )

  items = response['Items']
  print(items)

def scan(tableName):
  from boto3.dynamodb.conditions import Key, Attr

  table = dynamodb.Table(tableName)

  response = table.scan(
    FilterExpression=Attr('age').gt(23)
  )

  items = response['Items']
  print(items)

  len(items)
  for x in range(len(items)): 
    items[x]['username']

def query_filter(tableName):
  from boto3.dynamodb.conditions import Key, Attr

  table = dynamodb.Table(tableName)

  response = table.scan(
    FilterExpression=Attr('first_name').begins_with('r') & Attr('account_type').eq('librarian')
  )

  items = response['Items']
  print(items)


# Comment/uncomment below to play with the different functions
#create_table('staff')

#put_item('staff')
#get_item('staff')
#update_item('staff')
#delete_item('staff')

#batch_write('staff')

#create_multiple_items('staff', 100)

#query('staff')
#scan('staff')
#query_filter('staff')

#delete_table('staff')