How to work with AWS Lambda, DynamoDB and Python

Hello dear friends!

Today I wanna show you the connection between AWS tools: Lambda, API Gateway, and DynamoDB, with Python.

So first of all we need to know how it can work,

What is AWS Lambda?

AWS Lambda is a serverless app that allows you to build API applications (in most cases) without building or making a start build new machine server like EC2, from scratch, you pay as you go, which means the payment is dependent on your use (request and more.), so in this way, you don’t need to buy large server if you don’t sure you need it, or small server if you need more, also when your build a machine you when the instance still running you pay.

What is DynamoDB?

DynamoDB is a database with the NoSQL (JSON Bases like MongoDB) architecture on AWS, which means you don’t need to send data out on your server, it’s more secure and easy to use with AWS Products and Services.

What is API Gateway

You can think about API Gateway as a Gate of your AWS account to outside of the world with options to lock who can in and who doesn’t, you can build the API URL with authorization, password, patch, and more.

So let’s see how all these products work together:

Go to AWS account, lambda, and make a new lambda function with Python, then make a new trigger of API Gateway, and make a new DynamoDB Database,

In your lambda function add this files.

your-lambda-package
├── lambda_function.py
└── cdb.py

cdb.py

class qdb:
    def __init__(self,dynamoDBTable):
        self.client = boto3.resource('dynamodb')
        self.tableDynamoDB = dynamoDBTable
        self.table = self.client.Table(self.tableDynamoDB)
    def get(self,Id):
        response = self.table.get_item(
            Key={
                'id': Id
            }
        )
        if 'Item' in response:
            return True
        else:
            return False
    def create(self,Id):
        self.table.put_item(
                Item={
                    'id': Id
                }
            )
        return True

lambda_function.py

import json
import boto3
from cdb import qdb
def lambda_handler(event, context):
    
   
        customers = qdb('customers')
        
        if event and event['httpMethod']:
            Authorization = event['headers'].get('Authorization', None)
            if Authorization != None and event['headers']['Authorization'] == 'test':
                
                if event['httpMethod'] == "PUT":
                    bodyJson2Arr = json.loads(event['body'])
                    if bodyJson2Arr['tasktype'] == "create":
                        if bodyJson2Arr['data']['id']:
                            exist = customers.get(bodyJson2Arr['data']['id'])
                            message = exist
                            if not exist:
                                customers.create(bodyJson2Arr['data']['id'])
                                statusCode = 200
                                resTypeStatus = 'OK'
                                message = 'New User Id adding successfully'
                            else:
                                statusCode = 302
                                resTypeStatus = 'OK'
                                message = 'Sorry but this User alredy exist'
                        else:
                            statusCode = 419
                            resTypeStatus = 'MISSING ARGUMENTS'
                            message = 'Sorry but not found any Id in your request'
                    else:
                        statusCode = 419
                        resTypeStatus = 'MISSING ARGUMENTS'
                        message = 'Sorry but not found any Create in your request'
                        
                elif event['httpMethod'] == 'GET':
                    id = event['multiValueQueryStringParameters'].get('id', None)
                    if id == None:
                        statusCode = 419
                        resTypeStatus = 'MISSING ARGUMENTS'
                        message = 'Sorry but not found any Id in your request'
                    else:    
                        if customers.get(event['multiValueQueryStringParameters']['id'][0]):
                            statusCode = 302
                            resTypeStatus = 'OK'
                            message = 'This User exist'
                        else:
                            statusCode = 404
                            resTypeStatus = 'OK'
                            message = 'This User not found ):'
                else:
                    statusCode = 405
                    message= 'This method type is not currently supported.'
                    resTypeStatus = 'METHOD NOT ALLOWED'
            else:
                statusCode = 403
                message= 'Failed to pass authorization.'
                resTypeStatus = 'Forbidden'
    
        jsonDataBody = {'code': statusCode, 'status': resTypeStatus,'message': message}
    
        return {
            'headers': {
                "content-type":"application/json"
            },
            'statusCode': statusCode,
            'body': json.dumps(jsonDataBody)
        }

I also put the class file and the full code of the Lambda function

Get customersDB Class To Your Code and put your DynamoDB Tablecustomers = customersDB(‘MyTable’)

API Examples

Header Authorization

Authorization: test

GET – Check Id

URL

https://s73edk94da.execute-api.eu-west-1.amazonaws.com/default/server

Params

id: MyId

Example Request:

https://s73edk94da.execute-api.eu-west-1.amazonaws.com/default/server?id=MyId

Example Response

{
    "code": 302,
    "status": "OK",
    "message": "This User exist"
}

PUT – Add New

Example Request

URL

https://s73edk94da.execute-api.eu-west-1.amazonaws.com/default/server

Body

{
  "tasktype": "create",
  "data": {
    "id": "MyNewId"
  }
}

Example Response

{
    "code": 200,
    "status": "OK",
    "message": "New User Id adding successfully"
}

• GitHub

https://github.com/neryaelul/AWS-Lambda-Python-to-DynamoDB