DynamoDB Examples with output for blog post: https://blog.richardkeller.net/dynamodb-basics-for-your-next-serverless-project/

examples.md

DynamoDB Examples

Create DynamoDB Table

def create_user_table():
    """Create the user table."""
    try:
        result = client.create_table(
            TableName='user_table',
            BillingMode='PAY_PER_REQUEST',
            AttributeDefinitions=[
                {
                    'AttributeName': 'char',
                    'AttributeType': 'S'
                },
                {
                    'AttributeName': 'email',
                    'AttributeType': 'S'
                }
            ],
            KeySchema=[
                {
                    'AttributeName': 'email',
                    'KeyType': 'HASH'
                },
                {
                    'AttributeName': 'char',
                    'KeyType': 'RANGE'
                }
            ]
        )
        return result
    except Exception as e:
        return str(e)

create_user

Doc: Create a new user in DynamoDB.

Source:

def create_user(email):
    """Create a new user in DynamoDB."""
    result = user_table.put_item(Item={
        'email': email,
        'char': email[:1],
        'human_date': str(datetime.utcnow()),
        'created_at': str(int(time.time()))
    })
    return result

Result:

{
    "ResponseMetadata": {
        "RequestId": "8R47UH037A0K8H0EGKI98NOJE3VV4KQNSO5AEMVJF66Q9ASUAAJG",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "server": "Server",
            "date": "Sun, 18 Aug 2019 21:09:33 GMT",
            "content-type": "application/x-amz-json-1.0",
            "content-length": "2",
            "connection": "keep-alive",
            "x-amzn-requestid": "8R47UH037A0K8H0EGKI98NOJE3VV4KQNSO5AEMVJF66Q9ASUAAJG",
            "x-amz-crc32": "2745614147"
        },
        "RetryAttempts": 0
    }
}

index

Doc: Select all users from the table.

Source:

def index():
    """Select all users from the table."""
    users = user_table.scan()
    return users

Result:

{
    "Items": [
        {
            "created_at": "1566162573",
            "char": "r",
            "email": "[email protected]",
            "human_date": "2019-08-18 21:09:33.147595"
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "ResponseMetadata": {
        "RequestId": "80FUP7E3N3PRUK106AO386PUC3VV4KQNSO5AEMVJF66Q9ASUAAJG",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "server": "Server",
            "date": "Sun, 18 Aug 2019 21:09:33 GMT",
            "content-type": "application/x-amz-json-1.0",
            "content-length": "180",
            "connection": "keep-alive",
            "x-amzn-requestid": "80FUP7E3N3PRUK106AO386PUC3VV4KQNSO5AEMVJF66Q9ASUAAJG",
            "x-amz-crc32": "2972745174"
        },
        "RetryAttempts": 0
    }
}

read_user

Doc: Get a user from the table by email.

Source:

def read_user(email):
    """Get a user from the table by email."""
    result = user_table.query(
        KeyConditionExpression=Key('email').eq(email)
    )
    return result

Result:

{
    "Items": [
        {
            "created_at": "1566162573",
            "char": "r",
            "email": "[email protected]",
            "human_date": "2019-08-18 21:09:33.147595"
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "ResponseMetadata": {
        "RequestId": "UUJ2SSN1ISPL1R85H7KMKIUA1VVV4KQNSO5AEMVJF66Q9ASUAAJG",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "server": "Server",
            "date": "Sun, 18 Aug 2019 21:09:33 GMT",
            "content-type": "application/x-amz-json-1.0",
            "content-length": "180",
            "connection": "keep-alive",
            "x-amzn-requestid": "UUJ2SSN1ISPL1R85H7KMKIUA1VVV4KQNSO5AEMVJF66Q9ASUAAJG",
            "x-amz-crc32": "2972745174"
        },
        "RetryAttempts": 0
    }
}

update_user

Doc: Update user information.

Source:

def update_user(email):
    """Update user information."""
    result = user_table.update_item(
        Key={
            'email': email,
            'char': email[:1]
        },
        ExpressionAttributeValues={
            ':city': 'Tulsa',
            ':updated_at': str(time.time())
        },
        UpdateExpression='SET city = :city, '
                         'updated_at = :updated_at',
        ReturnValues='ALL_NEW',
    )
    return result

Result:

{
    "Attributes": {
        "city": "Tulsa",
        "updated_at": "1566162573.5953672",
        "char": "r",
        "created_at": "1566162573",
        "email": "[email protected]",
        "human_date": "2019-08-18 21:09:33.147595"
    },
    "ResponseMetadata": {
        "RequestId": "TT43NU6801F6LP0B2JVAELFAUNVV4KQNSO5AEMVJF66Q9ASUAAJG",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "server": "Server",
            "date": "Sun, 18 Aug 2019 21:09:33 GMT",
            "content-type": "application/x-amz-json-1.0",
            "content-length": "217",
            "connection": "keep-alive",
            "x-amzn-requestid": "TT43NU6801F6LP0B2JVAELFAUNVV4KQNSO5AEMVJF66Q9ASUAAJG",
            "x-amz-crc32": "2746400021"
        },
        "RetryAttempts": 0
    }
}

delete_user

Doc: Delete user by email.

Source:

def delete_user(email):
    """Delete user by email."""
    result = user_table.delete_item(
        Key={
            'email': email,
            'char': email[:1]
        }
    )
    return result

Result:

{
    "ResponseMetadata": {
        "RequestId": "G9LQ19E924MTFMMJ3OAA4FBOVNVV4KQNSO5AEMVJF66Q9ASUAAJG",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "server": "Server",
            "date": "Sun, 18 Aug 2019 21:09:33 GMT",
            "content-type": "application/x-amz-json-1.0",
            "content-length": "2",
            "connection": "keep-alive",
            "x-amzn-requestid": "G9LQ19E924MTFMMJ3OAA4FBOVNVV4KQNSO5AEMVJF66Q9ASUAAJG",
            "x-amz-crc32": "2745614147"
        },
        "RetryAttempts": 0
    }
}