Creating DynamoDB tables locally from CDK stacks

readme.md

In a CDK project, I wanted the ability to create the DynamoDB tables defined in my stack in a locally running DynamoDB container so that I could run integration tests with a real DB. Unfortunately there doesn't seem to be an obvious way to do this.

If you're using the serverless framework, there is a plugin called serverless-dynamodb-local which achieves this for you with a single command. It hooks in to the CloudFormation building process, extracts the tables, and makes requests to the DynamoDB API to create them.

There doesn't seem to be a sanctioned way of hooking in to the CDK synthesizing process, however, running cdk synth does export the CloudFormation templates in cdk.out, which are easy enough to parse.

So the plan is simple:

1. Synthesize the stack(s) to create the CloudFormation output (as JSON), via cdk synth).
2. Read the templates and extract the tables from each stack, filtering on their resource type.
3. Convert each table in to a table creation request (by removing a few keys).
4. Create the tables via the DynamoDB API.

Python code for creating the tables is included below.

This can be executed as part of a GitHub actions job like this:

      - name: CDK Synth
        run: cdk synth

      - name: Start DynamoDB Local
        uses: rrainn/dynamodb-action@v4.0.0
        with:
          port: ${{ env.DYNAMODB_PORT }}
          cors: '*'

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Pip install
        run: pip install boto3

      - name: Create DynamoDB tables
        run: python ./.github/workflows/scripts/create_stack_dynamodb_tables.py

create_stack_dynamodb_tables.py

from glob import glob
import os
import json
import boto3

dynamo = boto3.resource('dynamodb', endpoint_url='http://localhost:{port}'.format(port=os.environ['DYNAMODB_PORT']))

dynamodb_table_resource_types = ['AWS::DynamoDB::GlobalTable', 'AWS::DynamoDB::Table']
dynamodb_table_create_keys = ['AttributeDefinitions', 'TableName', 'KeySchema', 'LocalSecondaryIndexes', 'GlobalSecondaryIndexes', 'BillingMode', 'ProvisionedThroughput', 'StreamSpecification', 'SSESpecification', 'Tags', 'TableClass', 'DeletionProtectionEnabled', 'ResourcePolicy']

stack_template_files = glob('cdk.out/*.template.json')
for stack_template_file in stack_template_files:
  print('> 👀 Reading {file}'.format(file=stack_template_file))
  with open(stack_template_file) as f:
    stack_template = json.load(f)
  stack_resources = list(stack_template['Resources'].values())
  dynamo_db_table_resources = [r for r in stack_resources if r['Type'] in dynamodb_table_resource_types]
  print('>> 📋 Tables: {tables}'.format(tables=[t['Properties']['TableName'] for t in dynamo_db_table_resources]))

  for dynamo_table_resource in dynamo_db_table_resources:
    print('>>> 🏗️ Creating: {table} - {table_json}'.format(table=dynamo_table_resource['Properties']['TableName'], table_json=json.dumps(dynamo_table_resource)))
    table_create_request = {k: v for k, v in dynamo_table_resource['Properties'].items() if k in dynamodb_table_create_keys}
    try:
      dynamo.create_table(**table_create_request)
      print('>>> ✅ Created!')
    except Exception as e:
      print('>>> ❌ Failed: {message}'.format(message=str(e)))