使用DynamoDB使用无服务器框架构建CRUD Python无服务器API。-DEV365 开发者社区

嗨，我叫Noble Mutuwa Mulaudzi，AWS DevOps工程师和Linux爱好者。

在本教程中，我们将学习如何使用Python和DynamoDB构建无服务器API，从而利用无服务器框架的功能。 API将为管理项目提供创建，阅读，更新和删除（CRUD）操作。我们还将演示如何使用Postman测试API。

建筑图

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

先决条件

要与本教程一起，您需要以下内容：

python 3.x安装在您的机器上。
node.js和NPM安装用于安装无服务器框架。
AWS帐户具有适当的权限，可以创建Lambda之类的资源

设置项目

1.通过运行以下命令来安装无服务器框架：

npm install -g serverless

2.为您的项目创建一个新目录并导航。

mkdir python_rest_api
cd python_rest_api

3.使用以下命令进行新的无服务器服务：

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

4.通过运行：
安装所需的python软件包

pip install boto3

为项目目录中的每个CRUD操作创建单独的Python文件：

create.py
delete.py
update.py
read_one.py
read_all.py

配置DynamoDB

1.打开server.yml文件并使用以下配置进行更新：

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

实施CRUD功能

2.打开create.py文件并使用以下代码进行更新：

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.打开delete.py文件并使用以下代码进行更新：

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.打开update.py文件并使用以下代码进行更新：

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.打开read_one.py文件并使用以下代码进行更新：

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.打开read_all.py文件并使用以下代码进行更新：

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

使用以下命令部署无服务器API：

serverless deploy

使用Postman测试API

打开邮递员并创建一个新请求
将请求URL设置为您进行无服务器Deloy时获得的每个CRUD操作的适当端点：

替换为您先前注意的实际API端点。

3.设置创建和更新操作的请求主体。例如：

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

4.提出请求并观察响应。

发布请求：

获取请求：

恭喜！您已经使用无服务器框架成功地使用DynamoDB构建了CRUD Python无服务器API。您还使用Postman测试了API。
随时探索和扩展此基础，以构建更复杂的无服务器API以适合您的特定要求。
总结了教程。我们已经介绍了如何使用DynamoDB创建CRUD Python无服务器API，并通过Postman

建筑图