Setting up a REST API in Python for DynamoDB

Dynamo DB is AWS's NoSQL offering in the vast set of managed databases as a service that they provide. Like most other services, it's fully serverless, flexible and easy to scale.

The Data Model

As we're working on NoSQL here, there's no real restriction on the structure of data. We can operate with key-value pairs as the attributes for each item in a table. Let's look at these keywords again.

Table - a fairly familiar term, it is essentially a collection of data, in this case, items. It is also the starting point of working with DynamoDB on the console.

Item - an entry in a table. You could consider it a row in an SQL-equivalent database.

Attribute - The datapoints that constitute an item. It could contain item-specific attributes, metadata, or virtually anything that can be associated with an item.

You could think of a JSON array as an equivalent to a table in DynamoDB. I'm sure things will get clearer as we create our own table.

Setting up the database

It's literally a piece of cake to create a new table in DynamoDB from the AWS Console. All you need is a name and a partition key, which is your primary key in this case. This will help you search for items in the table.

I'm creating a table for all the games that I've played, and I will rate them out of 10 :)

You can mess with the table directly from the console, let's try adding a new item to see what it looks like.

My first entry has to be my favourite RPG (role-playing) game - The Witcher 3. I will add a new attribute for rating and it's going to be a solid 9.8 from me :)

Setting up an API

Right, it's now time to write some Python code to do all of this without the GUI ;)

## app.py
from flask import Flask, jsonify, request
import boto3
from boto3.dynamodb.conditions import Key
import uuid  # Import uuid module for generating UUIDs

app = Flask(__name__)

# Initialize DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='ap-south-1')  # Replace with your region
## Do keep in mind to save your AWS credentials file in the root directory
table = dynamodb.Table('games')  # Replace with your table name

# Route to get all games
@app.route('/games', methods=['GET'])
def get_games():
    try:
        response = table.scan()
        games = response.get('Items', [])
        return jsonify({'games': games}), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

The beauty of Python is that you can setup a full-fledged API in just a few lines of code. This chunk of code is now sufficient for us to access the table and fetch the data from it. We use the scan function to fetch items from the games table.

You can start the app by using python3 app.py

And you can expect a response that looks like this when you curl for the /games endpoint.

Routes for creating and updating an entry

## app.py
from flask import Flask, jsonify, request
import boto3
from boto3.dynamodb.conditions import Key
import uuid  # Import uuid module for generating UUIDs

app = Flask(__name__)

# Initialize DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='ap-south-1')  # Replace with your region
## Do keep in mind to save your AWS credentials file in the root directory
table = dynamodb.Table('games')  # Replace with your table name

# Route to get all games
@app.route('/games', methods=['GET'])
def get_games():
    try:
        response = table.scan()
        games = response.get('Items', [])
        return jsonify({'games': games}), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

Here, we are using put_item to add an item to the table. For updating a record, we use the function update_item.

If you observe carefully, we are using UpdateExpression where we specify the attributes that we're updating. This allows us to control exactly which attribute gets changed and avoid accidental overwrites.

And to delete the record, you can go with something like this -

# Route to create a new game
@app.route('/games', methods=['POST'])
def create_game():
    try:
        game_data = request.get_json()
        name = game_data.get('name')
        rating = game_data.get('rating')
        hours = game_data.get('hours', 0)

        # Generate a random UUID for the new game
        id = str(uuid.uuid4())

        if not name or not rating:
            return jsonify({'error': 'Missing required fields'}), 400

        # Store the game in DynamoDB
        table.put_item(Item={'id': id, 'name': name, 'rating': rating, 'hours': hours})

        # Return the created game with the generated UUID
        created_game = {'id': id, 'name': name, 'rating': rating}
        return jsonify({'message': 'Game added successfully', 'game': created_game}), 201
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# Route to update an existing game
@app.route('/games/<int:id>', methods=['PUT'])
def update_game(id):
    try:
        game_data = request.get_json()
        name = game_data.get('name')
        rating = game_data.get('rating')
        hours = game_data.get('hours', 0)

        if not name and not rating:
            return jsonify({'error': 'Nothing to update'}), 400

        update_expression = 'SET '
        expression_attribute_values = {}

        if name:
            update_expression += ' #n = :n,'
            expression_attribute_values[':n'] = name
        if rating:
            update_expression += ' #r = :r,'
            expression_attribute_values[':r'] = rating
        if hours:
            update_expression += ' #h = :h,'
            expression_attribute_values[':h'] = hours

        update_expression = update_expression[:-1]  # remove trailing comma
        response = table.update_item(
            Key={'id': id},
            UpdateExpression=update_expression,
            ExpressionAttributeNames={'#n': 'name', '#r': 'rating', '#h': 'hours'},
            ExpressionAttributeValues=expression_attribute_values,
            ReturnValues='UPDATED_NEW'
        )
        updated_game = response.get('Attributes', {})
        return jsonify(updated_game), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 500

---

Well, there you have it, you just setup a REST API with CRUD Functionality for DynamoDB in a matter of minutes thanks to Python.

The above is the detailed content of Setting up a REST API in Python for DynamoDB. For more information, please follow other related articles on the PHP Chinese website!