DynamoDB TestContainers in Python

Let’s get started straight away, shall we ?

Use pip to install testcontainers package

pip install testcontainers

testcontainers-python library has direct support for some widely used Docker images like Selenium, MySql, Oracle, RabbitMQ etc. But they didn’t implemented any wrapper to run DyanmoDB as TestContainer. Hence I have used the Generic container to setup DynamoDB

DockerContainer class helps us create container using any Docker image

container = DockerContainer("amazon/dynamodb-local") \
        .with_bind_ports(8000, 8000)\
        .with_command("-jar DynamoDBLocal.jar -sharedDb")

Note: Unless we specify the-sharedDb via command, DynamoDB will start with default as-inMemory.In that case, all data will be written to memory.

Refer local usage notes for more information

Connecting to DynamoDB Local Docker instance

dynamo_client = boto3.client('dynamodb',
                             endpoint_url='http://localhost:8000',
                             region_name='local',
                             aws_access_key_id='key',
                             aws_secret_access_key='secret')

We can pass random values to aws_access_key_id & aws_secret_access_key

Here is the complete code

import unittest
import boto3 as boto3
from testcontainers.core.container import DockerContainer


class TestingSum(unittest.TestCase):

    def setUp(self):
        self.container = DockerContainer("amazon/dynamodb-local") \
            .with_bind_ports(8000, 8000).with_command("-jar DynamoDBLocal.jar -sharedDb").with_exposed_ports(8080)

    def test_dyanmodb_table_creation(self):
        with self.container:
            client = boto3.client('dynamodb', endpoint_url='http://localhost:8000', region_name='local',
                                  aws_access_key_id='key', aws_secret_access_key='secret')
            self.create_table()
            response = client.list_tables()
            self.assertEqual(response['TableNames'], ['my-database'])
            self.assertEqual(response['ResponseMetadata']['HTTPStatusCode'], 200)

    def create_table(self):
        dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:8000', region_name='local',
                                aws_access_key_id='key', aws_secret_access_key='secret')

        dynamodb.create_table(
            TableName='my-database',
            KeySchema=[
                {
                    'AttributeName': 'id_key',
                    'KeyType': 'HASH'
                },
                {
                    'AttributeName': 'sort_key',
                    'KeyType': 'RANGE'
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'id_key',
                    'AttributeType': 'S'
                },
                {
                    'AttributeName': 'sort_key',
                    'AttributeType': 'S'
                },
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 10,
                'WriteCapacityUnits': 10
            }
        )
        dynamodb.close()


if __name__ == '__main__':
    unittest.main()

Thats all folks !!