Create a REST API with API Gateway, Lambda, and DynamoDB!
This project demonstrates the practical implementation of serverless design principles while leveraging AWS services to deliver a modern, scalable API solution.
Use Cases
- Building a web or mobile application’s backend API.
- Creating microservices for processing data or handling user requests.
- Prototyping scalable startup projects.
Each component plays a specific role in this architecture, enabling seamless interaction between clients and data storage.
Prequisite
Steps to follow
- Create an empty DynamoDB table that your Lambda function will perform Create, Read, Update, and Delete (CRUD) operations on.
- Open the Tables page of the DynamoDB console.
- Choose Create table
- Under Table Details, do the following:
(a) Table name, enter employee_info.
(b) For Partition key, enter employee_id, and keep the data type set as String.
Under Table settings, keep the Default settings.
- Choose Create table.
2. Create a Lambda Function
- In the AWS Search bar, type lambda and click to open the Lamda console page.
- Follow the Create function image to create a lambda function.
- Attach the DynamoDB and CloudWatch roles to the api_function created as shown below.
3. Create a REST API using API Gateway: create the API Gateway REST API that you will use to invoke your Lambda function.
To create the API
- Open the API Gateway console.
- Choose Create API.
In the REST API box, choose Build.
Under API details, leave New API selected, and for API Name, enter Dy.
- Choose Create API.
- Create the following method for the employee under Resources: GET, POST, PATCH and DELETE.
- This is how your
/employeeendpoint should look after creating the method.
Copy the Deploy API link and save it for later use: https://z3gr824o4l.execute-api.us-east-1.amazonaws.com/workloadapi
4. Build the Lambda Function: Write your Lambda function to handle API requests
import json
import boto3
from botocore.exceptions import ClientError
from decimal import Decimal
from boto3.dynamodb.conditions import Key
# Initialize the DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
dynamodb_table = dynamodb.Table('employee_info')
status_check_path = '/status'
employee_path = '/employee'
employees_path = '/employees'
def lambda_handler(event, context):
print('Request event: ', event)
response = None
try:
http_method = event.get('httpMethod')
path = event.get('path')
if http_method == 'GET' and path == status_check_path:
response = build_response(200, 'Service is operational')
elif http_method == 'GET' and path == employee_path:
employee_id = event['queryStringParameters']['employeeid']
response = get_employee(employee_id)
elif http_method == 'GET' and path == employees_path:
response = get_employees()
elif http_method == 'POST' and path == employee_path:
response = save_employee(json.loads(event['body']))
elif http_method == 'PATCH' and path == employee_path:
body = json.loads(event['body'])
response = modify_employee(body['employeeId'], body['updateKey'], body['updateValue'])
elif http_method == 'DELETE' and path == employee_path:
body = json.loads(event['body'])
response = delete_employee(body['employeeId'])
else:
response = build_response(404, '404 Not Found')
except Exception as e:
print('Error:', e)
response = build_response(400, 'Error processing request')
return response
def get_employee(employee_id):
try:
response = dynamodb_table.get_item(Key={'employeeid': employee_id})
return build_response(200, response.get('Item'))
except ClientError as e:
print('Error:', e)
return build_response(400, e.response['Error']['Message'])
def get_employees():
try:
scan_params = {
'TableName': dynamodb_table.name
}
return build_response(200, scan_dynamo_records(scan_params, []))
except ClientError as e:
print('Error:', e)
return build_response(400, e.response['Error']['Message'])
def scan_dynamo_records(scan_params, item_array):
response = dynamodb_table.scan(**scan_params)
item_array.extend(response.get('Items', []))
if 'LastEvaluatedKey' in response:
scan_params['ExclusiveStartKey'] = response['LastEvaluatedKey']
return scan_dynamo_records(scan_params, item_array)
else:
return {'employees': item_array}
def save_employee(request_body):
try:
dynamodb_table.put_item(Item=request_body)
body = {
'Operation': 'SAVE',
'Message': 'SUCCESS',
'Item': request_body
}
return build_response(200, body)
except ClientError as e:
print('Error:', e)
return build_response(400, e.response['Error']['Message'])
def modify_employee(employee_id, update_key, update_value):
try:
response = dynamodb_table.update_item(
Key={'employeeid': employee_id},
UpdateExpression=f'SET {update_key} = :value',
ExpressionAttributeValues={':value': update_value},
ReturnValues='UPDATED_NEW'
)
body = {
'Operation': 'UPDATE',
'Message': 'SUCCESS',
'UpdatedAttributes': response
}
return build_response(200, body)
except ClientError as e:
print('Error:', e)
return build_response(400, e.response['Error']['Message'])
def delete_employee(employee_id):
try:
response = dynamodb_table.delete_item(
Key={'employeeid': employee_id},
ReturnValues='ALL_OLD'
)
body = {
'Operation': 'DELETE',
'Message': 'SUCCESS',
'Item': response
}
return build_response(200, body)
except ClientError as e:
print('Error:', e)
return build_response(400, e.response['Error']['Message'])
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
# Check if it's an int or a float
if obj % 1 == 0:
return int(obj)
else:
return float(obj)
# Let the base class default method raise the TypeError
return super(DecimalEncoder, self).default(obj)
def build_response(status_code, body):
return {
'statusCode': status_code,
'headers': {
'Content-Type': 'application/json'
},
'body': json.dumps(body, cls=DecimalEncoder)
}- Verify that everything is working properly by testing with the Deploy API link.
The successful implementation of this project demonstrates the power and flexibility of serverless architecture using API Gateway, AWS Lambda, and DynamoDB. By leveraging these services, we have built a highly scalable, cost-effective, and reliable REST API that eliminates the need for traditional server management. This approach not only reduces operational overhead but also enables faster development and deployment cycles, making it ideal for modern, cloud-native applications.
