Rafal Wilinski

Posted on Mar 5, 2022 • Originally published at dynobase.dev

DynamoDB in Python's Boto3 - Query Cheat Sheet with Examples

#aws #serverless #dynamodb

What is Boto3?

Boto3 is a Python library for AWS (Amazon Web Services), which helps interacting with their services including DynamoDB - you can think of it as DynamoDB Python SDK. It empowers developers to manage and create AWS resources and DynamoDB Tables and Items.

If you're looking for similar guide but for Node.js, you can find it here, for Java, for Rust, and for Golang / Go here.

Connecting Boto3 to DynamoDB

Connecting to DynamoDB with boto3 is simple if you want to do that using Access and Secret Key combination:

import boto3

client = boto3.client('dynamodb',
  aws_access_key_id='yyyy',
  aws_secret_access_key='xxxx',
  region_name='us-east-1')

Keep in mind that using access and secret keys is against best security practices, and you should instead use IAM roles/policies to interact with DynamoDB. Boto3, if ran on Lamba function or EC2 instance, will automatically consume IAM Role attached to it.

Create Table with Boto3

DynamoDB structures data in tables, so if you want to save some data to DynamoDB, first you need to create a table. You can do that using AWS Console, AWS CLI or using boto3, like this:

import boto3

dynamodb = boto3.resource('dynamodb', region_name='us-west-2')


table = dynamodb.create_table(
    TableName='Movies',
    KeySchema=[
        {
            'AttributeName': 'year',
            'KeyType': 'HASH'  #Partition key
        },
        {
            'AttributeName': 'title',
            'KeyType': 'RANGE'  #Sort key
        }
    ],
    AttributeDefinitions=[
        {
            'AttributeName': 'id',
            'AttributeType': 'N'
        },
        {
            'AttributeName': 'createdAt',
            'AttributeType': 'S'
        },

    ],
    ProvisionedThroughput={
        'ReadCapacityUnits': 10,
        'WriteCapacityUnits': 10
    }
)

print("Table status:", table.table_status)

Keep in mind that provisioning a table takes some before it's active. If you want to know when it's ready to be used, you can use waiter function.

import botocore.session

session = botocore.session.get_session()
dynamodb = session.create_client('dynamodb', region_name='us-east-1') # low-level client

waiter = dynamodb.get_waiter('table_exists')
waiter.wait(TableName="my-table-name")

Boto3 Get All Items aka Scan

To get all items from DynamoDB table, you can use Scan operation. The problem is that Scan has 1 MB limit on the amount of data it will return in a request, so we need to paginate through the results in a loop.

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.scan()
data = response['Items']

while 'LastEvaluatedKey' in response:
    response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
    data.extend(response['Items'])

You can apply FilterExpression attribute in order to filter the results like this:

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.scan(FilterExpression=Attr('country').eq('US') & Attr('city').eq('NYC'))
data = response['Items']

while 'LastEvaluatedKey' in response:
    response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
    data.extend(response['Items'])

Boto3 Get Item

To get a single item from DynamoDB using Partition Key (and Sort Key if using composite key), you can use GetItem operation.

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.get_item(Key={
  primaryKeyName: "ID-1",
  sortKeyName: "SORT_2"
})

Keep in mind to replace primaryKeyName and sortKeyName with actual keys from your table.

Boto3 Batch Get Item

If you want to retrieve multiple items identified by a key(s) in one call, use batch_get_item call with the following syntax:

import boto3

dynamodb = boto3.resource('dynamodb', region_name=region)

response = dynamodb.batch_get_item(
        RequestItems={
            'my-table': {
                'Keys': [
                    {
                        'id': 1
                    },
                    {
                        'id': 2
                    },
                ],
                'ConsistentRead': True
            }
        },
        ReturnConsumedCapacity='TOTAL'
    )

Keep in mind that batch_get_item is limited to 100 items and 16 MB of data.

Boto3 Put Item

To write a single item into the DynamoDB Table, use PutItem operation:

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.put_item(
    Item={
        'id': 1,
        'title': 'my-document-title',
        'content': 'some-content',
    }
)

Boto3 Query for a set of items

Alternative way to get a collection of items is the Query method. Query is much faster than Scan because it uses Indexes. It should be your preferred way to get a collection of items with the same partition key.

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.query(
    KeyConditionExpression=Key('id').eq(1)
)

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

Keep in mind that Query can return up to 1MB of data and you can also use FilterExpressions here to narrow the results on non-key attributes.

If you don't know how to construct your Query, use Dynobase with Query Code Generation feature which will automatically generate it for you.

Boto3 Update Item

DynamoDB update_item operation consists of three primary attributes:

Key - which object should be updated
ExpressionAttributeValues - map with new values
UpdateExpression - how these new values should be applied to the object in the table

They can be used like this:

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.update_item(
    Key={
        'id': '894673'
    },
    UpdateExpression='SET country = :newCountry",
    ExpressionAttributeValues={
        ':newCountry': "Canada"
    },
    ReturnValues="UPDATED_NEW"
)

Boto3 Conditionally Update Item

Moreover, you can also add a ConditionExpression parameter, which restricts the update logic only if the evaluated expression equals true.

import boto3

dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.update_item(
    Key={
        'id': '894673'
    },
    UpdateExpression='SET country = :newCountry",
    ConditionExpression='attribute_not_exists(deletedAt)' # Do not update if deleted
    ExpressionAttributeValues={
        ':newCountry': "Canada"
    },
    ReturnValues="UPDATED_NEW"
)

Boto3 Increment Item Attribute

Incrementing a Number value in DynamoDB item can be achieved in two ways:

Fetch item, update the value with code and send a Put request overwriting item
Using update_item operation.

While it might be tempting to use first method because Update syntax is unfriendly, I strongly recommend using second one because of the fact it's much faster (requires only one request) and atomic (imagine value updated by other client after you fetched item).

To do that using single update_item operation, use following syntax:

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.update_item(
    Key={
        'id': '777'
    },
    UpdateExpression='SET score.#s = score.#s + :val",
    ExpressionAttributeNames={
        "#s": "goals"
    },
    ExpressionAttributeValues={
        ':val': decimal.Decimal(1)
    },
    ReturnValues="UPDATED_NEW"
)

Boto3 Delete Item

Deleting a single item from DynamoDB table is similar to GetItem operation. Key argument accepts primary key and sort/range key if table has composite key.

import boto3
dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.table.delete_item(Key={
    primaryKeyName: primaryKeyValue
})

Boto3 Delete All Items

Unfortunately, there's no easy way to delete all items from DynamoDB just like in SQL-based databases by using DELETE FROM my-table;. To achieve the same result in DynamoDB, you need to query/scan to get all the items in a table using pagination until all items are scanned and then perform delete operation one-by-one on each record.

import boto3

dynamodb = boto3.resource('dynamodb', region_name=region)
table = dynamodb.Table('my-table')

with table.batch_writer() as batch:
     # Iterate through table until it's fully scanned
    while scan is None or 'LastEvaluatedKey' in scan:
        if scan is not None and 'LastEvaluatedKey' in scan:
            scan = table.scan(
                ProjectionExpression='yourPrimaryKey', # Replace with your actual Primary Key
                ExclusiveStartKey=scan['LastEvaluatedKey'],
            )
        else:
            scan = table.scan(ProjectionExpression='yourPrimaryKey')

        for item in scan['Items']:
            batch.delete_item(Key={'yourPrimaryKey': item['yourPrimaryKey']})

Fortunately, this is possible just with 3 clicks using Dynobase.

Boto3 Query with Sorting

Unfortunately, DynamoDB offers only one way of sorting the results on the database side - using the sort key. If your table does not have one, your sorting capabilities are limited to sorting items in application code after fetching the results. However, if you need to sort DynamoDB results on sort key descending or ascending, you can use following syntax:

import boto3

dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.query(
    ScanIndexForward=False # true = ascending, false = descending
)
data = response['Items']

Boto3 Query Pagination

Similar to Scan operation, Query returns results up to 1MB of items. If you need to fetch more records, you need to issue a second call to fetch the next page of results. If LastEvaluatedKey was present in response object, this table has more items like requested and another call with ExclusiveStartKey should be sent to fetch more of them:

import boto3

dynamodb = boto3.resource('dynamodb', region_name=region)

table = dynamodb.Table('my-table')

response = table.query()
data = response['Items']

# LastEvaluatedKey indicates that there are more results
while 'LastEvaluatedKey' in response:
    response = table.query(ExclusiveStartKey=response['LastEvaluatedKey'])
    data.update(response['Items'])

Using Boto3 with DynamoDB Local/Offline

If you need to use DynamoDB offline locally, you can use DynamoDB local distributed by AWS or DynamoDB from Localstack. Connecting to it is as easy as changing the endpoint parameter in boto3.resource call.

import boto3

# Use port 8000 for DynamoDB Local and 4569 for DynamoDB from LocalStack
dynamodb = boto3.resource('dynamodb',
                          region_name=region,
                          endpoint_url='http://localhost:8000')

# ... rest of your code

Learn more about running and using DynamoDB locally.

DEV Community