- 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"
}
Comments
Let's comment your feelings that are more than good