dynamo_injector.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
A script for quickly injecting records into DynamoDB.

Useful for testing apps that hang off a DynamoDB event stream.
"""

import itertools
import sys
import time

import boto3

dynamodb = boto3.resource('dynamodb')

if len(sys.argv) != 3:
    sys.exit('Usage: importer <table_name> <start>')

TABLE_NAME = sys.argv[1]
START = sys.argv[2]

table = dynamodb.Table(TABLE_NAME)

for i in itertools.count(int(START)):
    table.put_item(
        Item={
            'identifier': 'record_%d' % i
        }
    )
    time.sleep(1)