Integrating with DynamoDB

Most applications will need a database for storing persistent data, and DynamoDB is the flagship Aws NoSql database. It is the easiest to get setup and integrated with Serverless Functions. For more information about how to structure and use the database visit the Aws documentation.


Create a DynamoDB table for storing emails

The following code snippet shows and example of how you can set the attributes and keys for a DynamoDB table and create the table.

    
from cdev.aws.dynamodb import Table, AttributeDefinition, KeyDefinition, key_type, attribute_type myAttributes = [ AttributeDefinition("email", attribute_type.S), AttributeDefinition("date_created", attribute_type.S) ] myKeys = [ KeyDefinition("email", key_type.HASH), KeyDefinition("date_created", key_type.RANGE) ] EmailTable = Table("EmailRegistryTable", myAttributes, myKeys)


Write data from a Serverless Function

Start a new Cdev project and replace the code in the src/hello_world/resources.py file with the following code.

    
import json import os import time import boto3 from cdev.aws.dynamodb import Table, AttributeDefinition, KeyDefinition, key_type, attribute_type from cdev.aws.lambda_function import ServerlessFunction myAttributes = [ AttributeDefinition("email", attribute_type.S), AttributeDefinition("date_created", attribute_type.S) ] myKeys = [ KeyDefinition("email", key_type.HASH), KeyDefinition("date_created", key_type.RANGE) ] EmailTable = Table("EmailRegistryTable", myAttributes, myKeys) client = boto3.resource('dynamodb') table_name = os.environ.get("TABLENAME") @ServerlessFunction("email_adder", environment={"TABLENAME": EmailTable.output.table_name}, permissions=[EmailTable.available_permissions.READ_AND_WRITE_TABLE]) def add_email_handler(event, context): print(event) # Load the body of the request into a data obj data = event.get("body") first_name = data.get("first_name") last_name = data.get("last_name") email = data.get("email") now = int( time.time() ) print(f"ADDING: {first_name}; {last_name}; {email}") table = client.Table(table_name) table.put_item( Item={ 'email': email, 'first_name': first_name, 'last_name': last_name, 'date_created': str(now) } ) return { "statusCode": 200, "body": json.dumps({ "message": "Email Added", "success": True }), "headers":{ "Access-Control-Allow-Origin": "*", "mode": "no-cors" } } @ServerlessFunction("scan", environment={"TABLENAME": EmailTable.output.table_name}, permissions=[EmailTable.available_permissions.READ_AND_WRITE_TABLE]) def scan(event, context): table = client.Table(table_name) response=table.scan() response['Items'] print(response)
Deploy the resources.

cdev deploy

Execute the function.

cdev run function.execute hello_world_comp.email_adder --event-data "{\"body\":{\"first_name\":\"Paul\",\"last_name\":\"Atreides\",\"email\":\"Muaddib@dune.com\"}}"

You can run the function above a few times after making changes to the email address (and other fields if you wish, but the email is mandatory as it is the primary key), to add more entries to your database and then execute the scan function with the command below:

cdev run function.execute hello_world_comp.scan

Check the function logs

cdev run function.logs hello_world_comp.scan

You will see a list of items that are in your DynamoDB table in your terminal.