Building a CRUD Python Serverless API with DynamoDB using the Serverless Framework.
Tags: serverless, python, dynamodb
Published: 2023-06-04 15:29:19

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
_

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/46fythcyt7m07258baim.png)

- Source code: (https://github.com/Mutuwa99/python-crud-api)

### 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

```

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

```
mkdir python_rest_api
cd python_rest_api

```

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

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

```

4.Install the required Python packages by running:

```
pip install boto3

```

5. 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
```

### 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
```

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
```

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
```

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
```

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
```

### Deploy the Serverless API using the following command:

```
serverless deploy

```

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ucvciic2aifte4930uhu.png)

### 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 <API endpoint> 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"
}

```

4.Send the request and observe the responses.


### POST Request:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vp8a20pc6tu8pswhl64n.png)

### Get Request:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/2jgoxbw6du52ltwxabic.png)

- 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