Building a CRUD Python Serverless API with DynamoDB using the Serverless Framework.

0416

Hi,My name is Noble Mutuwa Mulaudzi, AWS DevOps Engineer and Linux enthusiast.

In this tutorial, we’ll learn how to build a serverless API using Python and DynamoDB, leveraging the power of the Serverless Framework. The API will provide Create, Read, Update, and Delete (CRUD) operations for managing items. We’ll also demonstrate how to test the API using Postman.

_

Architecture diagram

_

Prerequisites

To follow along with this tutorial, you’ll need the following:

  • Python 3.x installed on your machine.

  • Node.js and npm installed for installing the Serverless Framework.

  • An AWS account with appropriate permissions to create resources like Lambda

Setting up the Project

1.Install the Serverless Framework by running the following command: 

npm install -g serverless
npm install -g serverless
npm install -g serverless

Enter fullscreen mode Exit fullscreen mode

2.Create a new directory for your project and navigate into it. 

mkdir python_rest_api
cd python_rest_api
mkdir python_rest_api
cd python_rest_api
mkdir python_rest_api
cd python_rest_api

Enter fullscreen mode Exit fullscreen mode

3.Initialize a new Serverless service using the following command: 

serverless create --template aws-python3 --name python_rest_api
serverless create --template aws-python3 --name python_rest_api
serverless create --template aws-python3 --name python_rest_api

Enter fullscreen mode Exit fullscreen mode

4.Install the required Python packages by running: 

pip install boto3
pip install boto3
pip install boto3

Enter fullscreen mode Exit fullscreen mode

  1. Create separate Python files for each CRUD operation inside the project directory:

  • create.py

  • delete.py

  • update.py

  • read_one.py

  • read_all.py

Configuring DynamoDB

1.Open the serverless.yml file and update it with the following configuration: 

service: my-serverless-api
provider:
  name: aws
  runtime: python3.8
functions:
  create:
    handler: create.create
    events:
      - http:
          path: items
          method: post
  readAll:
    handler: read_all.read_all
    events:
      - http:
          path: items
          method: get
  readOne:
    handler: read_one.read_one
    events:
      - http:
          path: items/{id}
          method: get
  update:
    handler: update.update
    events:
      - http:
          path: items/{id}
          method: put
  delete:
    handler: delete.delete
    events:
      - http:
          path: items/{id}
          method: delete
resources:
  Resources:
    ItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: Items
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
service: my-serverless-api

provider:
  name: aws
  runtime: python3.8

functions:
  create:
    handler: create.create
    events:
      - http:
          path: items
          method: post

  readAll:
    handler: read_all.read_all
    events:
      - http:
          path: items
          method: get

  readOne:
    handler: read_one.read_one
    events:
      - http:
          path: items/{id}
          method: get

  update:
    handler: update.update
    events:
      - http:
          path: items/{id}
          method: put

  delete:
    handler: delete.delete
    events:
      - http:
          path: items/{id}
          method: delete

resources:
  Resources:
    ItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: Items
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
service: my-serverless-api

provider:
  name: aws
  runtime: python3.8

functions:
  create:
    handler: create.create
    events:
      - http:
          path: items
          method: post

  readAll:
    handler: read_all.read_all
    events:
      - http:
          path: items
          method: get

  readOne:
    handler: read_one.read_one
    events:
      - http:
          path: items/{id}
          method: get

  update:
    handler: update.update
    events:
      - http:
          path: items/{id}
          method: put

  delete:
    handler: delete.delete
    events:
      - http:
          path: items/{id}
          method: delete

resources:
  Resources:
    ItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: Items
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1

Enter fullscreen mode Exit fullscreen mode

Implementing the CRUD Functions

2.Open the create.py file and update it with the following code: 

import json
import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')
def create(event, context):
    data = json.loads(event['body'])
    item_id = data['id']
    item_name = data['name']
    item_price = data['price']
    try:
        table.put_item(
            Item={
                'id': item_id,
                'name': item_name,
                'price': item_price
            }
        )
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item created successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def create(event, context):
    data = json.loads(event['body'])
    item_id = data['id']
    item_name = data['name']
    item_price = data['price']

    try:
        table.put_item(
            Item={
                'id': item_id,
                'name': item_name,
                'price': item_price
            }
        )
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item created successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def create(event, context):
    data = json.loads(event['body'])
    item_id = data['id']
    item_name = data['name']
    item_price = data['price']

    try:
        table.put_item(
            Item={
                'id': item_id,
                'name': item_name,
                'price': item_price
            }
        )
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item created successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response

Enter fullscreen mode Exit fullscreen mode

3.Open the delete.py file and update it with the following code: 

import json
import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')
def delete(event, context):
    item_id = event['pathParameters']['id']
    try:
        table.delete_item(Key={'id': item_id})
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item deleted successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def delete(event, context):
    item_id = event['pathParameters']['id']

    try:
        table.delete_item(Key={'id': item_id})
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item deleted successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def delete(event, context):
    item_id = event['pathParameters']['id']

    try:
        table.delete_item(Key={'id': item_id})
        response = {
            'statusCode': 200,
            'body': json.dumps({'message': 'Item deleted successfully'})
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response

Enter fullscreen mode Exit fullscreen mode

4.Open the update.py file and update it with the following code: 

import json
import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')
def update(event, context):
    item_id = event['pathParameters']['id']
    data = json.loads(event['body'])
    item_name = data['name']
    item_price = data['price']
    try:
        response = table.update_item(
            Key={'id': item_id},
            UpdateExpression='set #name = :n, #price = :p',
            ExpressionAttributeNames={'#name': 'name', '#price': 'price'},
            ExpressionAttributeValues={':n': item_name, ':p': item_price},
            ReturnValues='UPDATED_NEW'
        )
        updated_item = response['Attributes']
        response = {
            'statusCode': 200,
            'body': json.dumps(updated_item)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def update(event, context):
    item_id = event['pathParameters']['id']
    data = json.loads(event['body'])
    item_name = data['name']
    item_price = data['price']

    try:
        response = table.update_item(
            Key={'id': item_id},
            UpdateExpression='set #name = :n, #price = :p',
            ExpressionAttributeNames={'#name': 'name', '#price': 'price'},
            ExpressionAttributeValues={':n': item_name, ':p': item_price},
            ReturnValues='UPDATED_NEW'
        )
        updated_item = response['Attributes']
        response = {
            'statusCode': 200,
            'body': json.dumps(updated_item)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def update(event, context):
    item_id = event['pathParameters']['id']
    data = json.loads(event['body'])
    item_name = data['name']
    item_price = data['price']

    try:
        response = table.update_item(
            Key={'id': item_id},
            UpdateExpression='set #name = :n, #price = :p',
            ExpressionAttributeNames={'#name': 'name', '#price': 'price'},
            ExpressionAttributeValues={':n': item_name, ':p': item_price},
            ReturnValues='UPDATED_NEW'
        )
        updated_item = response['Attributes']
        response = {
            'statusCode': 200,
            'body': json.dumps(updated_item)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response

Enter fullscreen mode Exit fullscreen mode

5.Open the read_one.py file and update it with the following code: 

import json
import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')
def read_one(event, context):
    item_id = event['pathParameters']['id']
    try:
        response = table.get_item(Key={'id': item_id})
        item = response.get('Item')
        if item:
            response = {
                'statusCode': 200,
                'body': json.dumps(item)
            }
        else:
            response = {
                'statusCode': 404,
                'body': json.dumps({'error': 'Item not found'})
            }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def read_one(event, context):
    item_id = event['pathParameters']['id']

    try:
        response = table.get_item(Key={'id': item_id})
        item = response.get('Item')
        if item:
            response = {
                'statusCode': 200,
                'body': json.dumps(item)
            }
        else:
            response = {
                'statusCode': 404,
                'body': json.dumps({'error': 'Item not found'})
            }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def read_one(event, context):
    item_id = event['pathParameters']['id']

    try:
        response = table.get_item(Key={'id': item_id})
        item = response.get('Item')
        if item:
            response = {
                'statusCode': 200,
                'body': json.dumps(item)
            }
        else:
            response = {
                'statusCode': 404,
                'body': json.dumps({'error': 'Item not found'})
            }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response

Enter fullscreen mode Exit fullscreen mode

6.Open the read_all.py file and update it with the following code: 

import json
import boto3
from botocore.exceptions import ClientError
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')
def read_all(event, context):
    try:
        response = table.scan()
        items = response['Items']
        response = {
            'statusCode': 200,
            'body': json.dumps(items)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }
    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def read_all(event, context):
    try:
        response = table.scan()
        items = response['Items']
        response = {
            'statusCode': 200,
            'body': json.dumps(items)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response
import json
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Items')

def read_all(event, context):
    try:
        response = table.scan()
        items = response['Items']
        response = {
            'statusCode': 200,
            'body': json.dumps(items)
        }
    except ClientError as e:
        response = {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

    return response

Enter fullscreen mode Exit fullscreen mode

Deploy the Serverless API using the following command: 

serverless deploy
serverless deploy
serverless deploy

Enter fullscreen mode Exit fullscreen mode

Testing the API using Postman

  1. Open Postman and create a new request

  2. Set the request URL to the appropriate endpoint for each CRUD operation that you got when you did serverless deloy :

  • Replace with the actual API endpoint you noted down earlier.

3.Set the request body for the create and update operations. For example: 

{
  "id": "1",
  "name": "Noble's Bill",
  "price": "R10 000"
}
{
  "id": "1",
  "name": "Noble's Bill",
  "price": "R10 000"
}
{
  "id": "1",
  "name": "Noble's Bill",
  "price": "R10 000"
}

Enter fullscreen mode Exit fullscreen mode

4.Send the request and observe the responses.

POST Request:

Get Request:

  • Congratulations! You have successfully built a CRUD Python serverless API with DynamoDB using the Serverless Framework. You’ve also tested the API using Postman.

  • Feel free to explore and expand upon this foundation to build more complex serverless APIs to suit your specific requirements.

  • That concludes the tutorial. We’ve covered how to create a CRUD Python serverless API with DynamoDB using the Serverless Framework and testing through postman

Article by Noble Mutuwa Mulaudzi

原文链接：Building a CRUD Python Serverless API with DynamoDB using the Serverless Framework.

展开阅读全文
© 版权声明
文章版权声明 1、本网站名称：拾光赋
2、本站永久网址：https://www.blogs.ink
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长QQ：805375623进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END
Python（EN）
# python# serverless# dynamodb
喜欢就支持一下吧
点赞6 分享
There are two ways of spreading light: to be the candle or the mirror that reflects it.
传递光亮有两种方式：成为一支蜡烛或当一面镜子
评论 抢沙发

请登录后发表评论

    暂无评论内容

文章目录
kity的头像-拾光赋
文章目录
今日剩余 57.7%
本周剩余 36.8%
本月剩余 65.3%
本年剩余 72.5%
最近评论
Vito的头像-拾光赋

Vito6个月前2

你就是我心中的那首忐忑，总是让我惊心动魄
Jason.Meng的头像-拾光赋钻石会员

Jason.Meng徽章-人气大使-拾光赋6个月前2

总是会在人生的不同阶段反复爱上大佬的文章~
每日一言
If you hold tight, how can a free hand to hug now?
你若将过去抱的太紧，怎么能腾出手来拥抱现在？