AWS LambdaのDynamoサンプルコード

dynamoTest.py

# APIGateway + Lambda + DynamoDB
# sample query
# {"tableName":"performance_test","operation":"read","payload": {"test_id":"aaaaaaa"}}
# {"tableName":"performance_test","operation":"delete","payload": {"test_id":"aaaaaaa"}}
# {"tableName":"performance_test","operation":"create","payload": {"test_id":"aaaaaaa","value":"aaaaaaaaa"}}

import boto3
import json

print('Loading function')


def lambda_handler(event, context):
    '''Provide an event that contains the following keys:

      - operation: one of the operations in the operations dict below
      - tableName: required for operations that interact with DynamoDB
      - payload: a parameter to pass to the operation being performed
    '''
    #print("Received event: " + json.dumps(event, indent=2))

    operation = event['operation']

    if 'tableName' in event:
        dynamo = boto3.resource('dynamodb').Table(event['tableName'])

    operations = {
        'create': lambda x: dynamo.put_item(Item=x),
        'read': lambda x: dynamo.get_item(Key=x),
        'update': lambda x: dynamo.update_item(x), # cannot execute
        'delete': lambda x: dynamo.delete_item(Key=x),
        'list': lambda x: dynamo.scan(Item=x), # cannot execute
        'echo': lambda x: x,
        'ping': lambda x: 'pong'
    }

    if operation in operations:
        return operations[operation](event['payload'])
    else:
        raise ValueError('Unrecognized operation "{}"'.format(operation))