Flask で APIGateway をシュミレート

ローカルにインストールしたDynamoDBに接続するAPIGatewayをシュミレートする方法です。
DynamoDB のインストールはこちら
ローカルのDynamoDB に、python3 で CRUD を行う
Cors の設定はこちら
Flask で Cors を使う

次のコマンドで Flask を起動する環境とします。

python welcome.py

次の API を作成します。引数はPOST で与えます。
http://localhost:5000/table_scan
http://localhost:5000/table_query

welcome.py

# -*- coding: utf-8 -*-
#
#   welcome.py
#
#                       Nov/07/2017
#
# ----------------------------------------------------------------------
import sys
import os

from flask import Flask
from flask_cors import CORS
from flask import jsonify
from flask import request
app = Flask(__name__)
CORS(app)

from table_scan import table_scan_exec
from table_query import table_query_exec
# ----------------------------------------------------------------------
@app.route("/")
def welcome():
    print("welcome")
    return app.send_static_file('index.html')
# ----------------------------------------------------------------------
@app.route('/table_scan' ,methods=['POST'])
def table_scan_proc():
    print("*** table_scan *** start ***")
    contents = []
#
    table_name = request.form['table_name']
#
    try:
        contents = table_scan_exec(table_name)
    except Exception as ee:
        sys.stderr.write("*** error *** table_scan_exec ***\n")
        sys.stderr.write(str(ee) + "\n")
#
    return jsonify(contents)

# ----------------------------------------------------------------------
@app.route('/table_query' ,methods=['POST'])
def table_query_proc():
    print("*** table_query *** start ***")
    table_name = "No table_name"
    key_query = "No key_query"
    key_value = "No value_query"
#
    contents = []
#
    table_name = request.form['table_name']
    key_query = request.form['key_query']
    value_query = request.form['value_query']
#
    try:
        contents = table_query_exec(table_name,key_query,value_query)
    except Exception as ee:
        sys.stderr.write("*** error *** table_query_exec ***\n")
        sys.stderr.write(str(ee) + "\n")
#
    return jsonify(contents)
# ----------------------------------------------------------------------
port = os.getenv('PORT', '5000')
if __name__ == '__main__':
    app.debug = True
    app.run(host='0.0.0.0',  port=int(port))
# ----------------------------------------------------------------------

table_scan.py

#! /usr/bin/python
# -*- coding: utf-8 -*-
#
#   table_scan.py
#
#                   Nov/07/2017
# --------------------------------------------------------------------
import  sys
import  boto3
from boto3.dynamodb.conditions import Key
#
#
from db_config import db_config_proc
from convert_structure import convert_structure_proc
# --------------------------------------------------------------------
def table_scan_exec(table_name):
    rvalue = []
    dynamodb = db_config_proc()
    table = dynamodb.Table(table_name)
    response = table.scan()
    rvalue = convert_structure_proc(response)
    return rvalue
# --------------------------------------------------------------------

table_query.py

# -*- coding: utf-8 -*-
#
#   table_query.py
#
#                   Nov/07/2017
# --------------------------------------------------------------------
import  sys
import  boto3
from boto3.dynamodb.conditions import Key
#
#
from db_config import db_config_proc
from convert_structure import convert_structure_proc
# --------------------------------------------------------------------
def table_query_exec(table_name,key_query,value_query):
    rvalue = []
    dynamodb = db_config_proc()
    table = dynamodb.Table(table_name)
    response = table.query(
        KeyConditionExpression=Key(key_query).eq(value_query)
        )
    rvalue = convert_structure_proc(response)
    return rvalue
# --------------------------------------------------------------------

db_config.py

# -*- coding: utf-8 -*-
#
#   db_config.py
#
#                   Nov/07/2017
# ---------------------------------------------------------------
import  sys
import  boto3
# ---------------------------------------------------------------
def db_config_proc():
    dynamodb = boto3.resource('dynamodb',endpoint_url="http://localhost:8000")
#
    return dynamodb
#
# ---------------------------------------------------------------

convert_structure.py

# -*- coding: utf-8 -*-
#
#   convert_structure.py
#
#                   Nov/07/2017
# --------------------------------------------------------------------
import  sys
import  boto3
import  decimal
# --------------------------------------------------------------------
def convert_structure_proc(response):
    array_aa = []
    for unit in response['Items']:
        str_out = ""
        unit_aa = {}
        for key in unit:
            str_out += str(unit[key]) + '\t'
            if type(unit[key]) == str:
                unit_aa[key] = unit[key]
            elif type(unit[key]) == decimal.Decimal:
                str_tmp = str(unit[key])
                unit_aa[key] = int(str_tmp)
            elif type(unit[key]) == dict:
                value = ''
                for vv in unit[key].values():
                    value = vv
                unit_aa[key] = value
            elif type(unit[key]) == bool:
                unit_aa[key] = str(unit[key])
            else:
                sys.stderr.write("key = " + key + '\n')
                sys.stderr.write(str(type(unit[key])) + '\n')
                unit_aa[key] = unit[key]
        array_aa.append(unit_aa)
#
    return array_aa
# --------------------------------------------------------------------