AWS ChaliceとPynamoDBでCRUD APIを作る

2021年3月11日

PynamoDBはDynamoDBをモデルクラスに抽象化して扱えるライブラリです。

AWS Chaliceに導入してCRUD APIを作ってみます。

1. ライブラリ導入準備
2. 実装
- 2.1. モデルの作成
- 2.2. CRUD APIの実装

ライブラリ導入準備

「requirements.txt」を作成し「pip3 install -r requirements.txt」します。

pynamodb

boto3を使っていればデプロイ時にChaliceが動的にIAMポリシーに権限を割当ててくれますが、PynamoDBを使っていると割り当たらないので、静的に割り当てるよう「/.chalice/policy.json」を作成します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:*"
            ],
            "Resource": [
                "*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:*:logs:*:*:*"
        }
    ]
}

「/.chalice/config.json」に作成したポリシーを使うよう設定します。

{
  "version": "2.0",
  "app_name": "pynamodb-crud-api",
  "stages": {
    "dev": {
      "api_gateway_stage": "api",
      "autogen_policy": false,
      "iam_policy_file": "policy.json"
    }
  }
}

実装

モデルの作成

「/chalicelib/」ディレクトリを作成し中に「__init__.py」とモデルのクラスを作成します。

from pynamodb.models import Model
from pynamodb.attributes import (
    UnicodeAttribute,
    NumberAttribute,
    ListAttribute,
)


class UserModel(Model):

    class Meta:
        table_name = "USERS"
        region = 'ap-northeast-1'

    id = UnicodeAttribute(hash_key=True)
    name = UnicodeAttribute(null=False)
    age = NumberAttribute(null=False)
    hobbies = ListAttribute(null=True)

CRUD APIの実装

from chalice import Chalice
from chalice import (
    ChaliceViewError,
    NotFoundError,
)

from chalicelib.UserModel import UserModel

import logging
import json
import uuid

logger = logging.getLogger()
logger.setLevel(logging.INFO)

app = Chalice(app_name='pynamodb-crud-api')


@app.route('/user', methods=['POST'],
           content_types=['application/json'])
def create_user():
    request = app.current_request
    logger.info(json.dumps(request.json_body))

    try:
        if not UserModel.exists():
            UserModel.create_table(
                read_capacity_units=1, write_capacity_units=1, wait=True)

        id = str(uuid.uuid4())

        user = UserModel(id)
        user.name = request.json_body['name']
        user.age = int(request.json_body['age'])
        if 'hobbies' in request.json_body:
            user.hobbies = request.json_body['hobbies']
        user.save()

    except Exception as e:
        logger.error(e)
        raise ChaliceViewError(str(e))

    return {
        'id': user.id,
        'name': user.name,
        'age': user.age,
        'hobbies': user.hobbies,
    }


@app.route('/users', methods=['GET'])
def get_users():
    try:
        users = UserModel.scan()
        response = []
        for user in users:
            response.append({
                'id': user.id,
                'name': user.name,
                'age': user.age,
                'hobbies': user.hobbies,
            })

    except Exception as e:
        logger.error(e)
        raise ChaliceViewError(str(e))

    return response


@app.route('/user/{id}', methods=['GET'])
def get_user_by_id(id):
    try:
        user = UserModel.get(id)

    except UserModel.DoesNotExist:
        logger.warn(id + ' is not found.')
        raise NotFoundError(id + ' is not found.')

    except Exception as e:
        logger.error(e)
        raise ChaliceViewError(str(e))

    return {
        'id': user.id,
        'name': user.name,
        'age': user.age,
        'hobbies': user.hobbies,
    }


@app.route('/user/{id}', methods=['PUT'],
           content_types=['application/json'])
def update_user_by_id(id):
    request = app.current_request
    logger.info(json.dumps(request.json_body))

    name = request.json_body['name']
    age = int(request.json_body['age'])
    if 'hobbies' in request.json_body:
        hobbies = request.json_body['hobbies']
    else:
        hobbies = None

    try:
        user = UserModel.get(id)
        user.update(actions=[
            UserModel.name.set(name),
            UserModel.age.set(age),
            UserModel.hobbies.set(hobbies),
        ])

    except UserModel.DoesNotExist:
        logger.warn(id + ' is not found.')
        raise NotFoundError(id + ' is not found.')

    except Exception as e:
        logger.error(e)
        raise ChaliceViewError(str(e))

    return {
        'id': user.id,
        'name': user.name,
        'age': user.age,
        'hobbies': user.hobbies,
    }


@app.route('/user/{id}', methods=['DELETE'])
def delete_user_by_id(id):
    try:
        user = UserModel.get(id)
        user.delete()

    except UserModel.DoesNotExist:
        logger.warn(id + ' is not found.')
        raise NotFoundError(id + ' is not found.')

    except Exception as e:
        logger.error(e)
        raise ChaliceViewError(str(e))

    return {
        'id': user.id,
        'name': user.name,
        'age': user.age,
        'hobbies': user.hobbies,
    }

あとはデプロイしてリクエストすればCRUDの処理が確認できます。