MySQLを利用したFlask RESTful API作成メモ #Python

PythonフレームワークFlaskとMySQLを利用したRESTful API(登録、取得のみ)の作成方法についてメモする。

準備

検証用MySQLコンテナ準備

docker-composeを使用して、MySQLコンテナを起動する。
- 以下のdocker-compose.ymlファイルの配置しているフォルダでdocker-compose upコマンドを実行する。
- 127.0.0.1:3307でMySQLコンテナが起動する。

version: "3"
services:
  db:
    image: mysql:5.7
    container_name: db
    environment:
      MYSQL_ROOT_PASSWORD: rootpass
      MYSQL_DATABASE: sample_db
      MYSQL_USER: mysqluser
      MYSQL_PASSWORD: mysqlpass
    volumes:
      - ./db/data:/var/lib/mysql
    ports:
      - 3307:3307
    command: --port 3307

以下のSQLを上記DBに実行し、テスト用テーブルusersを作成する。

  CREATE TABLE `sample_db`.`users` (
    `id` INT NOT NULL AUTO_INCREMENT,
    `name` VARCHAR(45) NOT NULL,
    `email_address` VARCHAR(128) NOT NULL,
    PRIMARY KEY (`id`),
    UNIQUE INDEX `id_UNIQUE` (`id` ASC),
    UNIQUE INDEX `name_UNIQUE` (`name` ASC))
  ENGINE = InnoDB
  DEFAULT CHARACTER SET = utf8mb4;

モジュールインストール

pip install flask, SQLAlchemy, Flask-SQLAlchemy, marshmallow, marshmallow-sqlalchemy, flask-marshmallow, pymysql

DB接続設定

検証用MySQLコンテナとの接続設定

from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://mysqluser:mysqlpass@127.0.0.1:3307/sample_db'
db = SQLAlchemy(app)

モデル＋コントローラー

検証用コード全体app.py

from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow_sqlalchemy import ModelSchema
import json
app = Flask(__name__)
## DB接続設定 
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://mysqluser:mysqlpass@127.0.0.1:3307/sample_db'
db = SQLAlchemy(app)
ma = Marshmallow()

## モデル
class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(45), nullable=False)
    email_address = db.Column(db.String(128), nullable=True)

    def __repr__(self):
        return "<User(name='%s',email_address='%s')>" % (self.name, self.email_address)

    # ユーザー情報登録
    def registUser(user):
        record = User(
            name=user['name'],
            email_address=user['email_address'],
        )
        db.session.add(record)
        db.session.commit()
        return user

    # ユーザー情報取得
    def getUserByUserId(uid):
        user = User.query.get(uid)
        return user


class UserSchema(ma.ModelSchema):
    class Meta:
        model = User
        fields = ('id', 'name', 'email_address')

## コントローラー
# ユーザー情報登録
@app.route('/users/', methods=['POST'])
def registUser():
    reqData = json.dumps(request.json)
    userData = json.loads(reqData)
    user = User.registUser(userData)
    user_schema = UserSchema()
    return jsonify(user_schema.dump(user))

# ユーザー情報取得
@app.route('/users/<string:uid>/', methods=['GET'])
def getUser(uid):
    user = User.getUserByUserId(uid)
    user_schema = UserSchema()
    if user == {}:
        return jsonify({})
    return make_response(jsonify(user_schema.dump(user)))


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

動作確認

localhost:5000で起動する。

python app.py

ユーザー情報登録

リクエスト

POST /users/ HTTP/1.1
Host: localhost:5000
Content-Type: application/json
Content-Length: 70

{
    "name":"test3 user",
    "email_address":"test3@example.com"
}

レスポンス

{
    "email_address": "test3@example.com",
    "name": "test3 user"
}

ユーザー情報取得

リクエスト

GET /users/3/ HTTP/1.1
Host: localhost:5000
Content-Type: application/json

レスポンス

{
    "email_address": "test3@example.com",
    "id": 3,
    "name": "test3 user"
}

参考情報

Create a CRUD Restful Service API using Flask + Mysql [in 7 minutes!]