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.
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)
Start a new Cdev project and replace the code in the src/hello_world/resources.py
file with the following code.
Deploy the resources.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)
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.