在 Python 中为 DynamoDB 设置 REST API

Dynamo DB 是 AWS 在大量托管数据库中提供的 NoSQL 产品，作为其提供的服务。与大多数其他服务一样，它完全无服务器、灵活且易于扩展。

数据模型

由于我们正在研究 NoSQL，因此对数据结构没有真正的限制。我们可以使用键值对作为表中每个项的属性进行操作。让我们再看看这些关键字。

表 - 一个相当熟悉的术语，它本质上是数据的集合，在本例中为项目。这也是在控制台上使用 DynamoDB 的起点。

Item - 表中的条目。您可以将其视为 SQL 等效数据库中的一行。

属性 - 构成项目的数据点。它可以包含特定于项目的属性、元数据或几乎任何可以与项目关联的内容。

您可以将 JSON 数组视为 DynamoDB 中的表的等效项。我相信当我们创建自己的表格时，事情会变得更加清晰。

设置数据库

从 AWS 控制台在 DynamoDB 中创建新表简直就是小菜一碟。您所需要的只是一个名称和一个分区键，在本例中这是您的主键。这将帮助您搜索表中的项目。

我正在为我玩过的所有游戏创建一个表格，我会给它们评分（满分 10 分）:)

您可以直接从控制台弄乱表格，让我们尝试添加一个新项目看看它是什么样子。

我的第一个作品必须是我最喜欢的 RPG（角色扮演）游戏 - 《巫师 3》。我将为评分添加一个新属性，这将是我的 9.8 分:)

设置 API

对了，现在是时候编写一些 Python 代码来在没有 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)

Python 的美妙之处在于您只需几行代码即可设置完整的 API。现在，这段代码足以让我们访问该表并从中获取数据。我们使用扫描功能从游戏桌上获取项目。

您可以使用 python3 app.py 启动应用程序

当您卷曲 /games 端点时，您可以期待如下所示的响应。

创建和更新条目的路由

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

在这里，我们使用 put_item 将项目添加到表中。为了更新记录，我们使用函数 update_item。

如果您仔细观察，我们正在使用 UpdateExpression 来指定要更新的属性。这使我们能够准确控制哪个属性被更改并避免意外覆盖。

要删除记录，您可以使用类似的操作 -

# 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

---

好了，你已经完成了，借助 Python，您只需几分钟即可为 DynamoDB 设置具有 CRUD 功能的 REST API。

以上是在 Python 中为 DynamoDB 设置 REST API的详细内容。更多信息请关注PHP中文网其他相关文章！