Flask SQLAlchemy で REST API を作ってみた
ORMとは
ORM（Object-Relational Mapping）とは、データベースとオブジェクト指向プログラミング言語とのマッピングを行うことを指します。
ただこれだけ聞くと、イメージしづらいですが、簡単に言うと、SQL文を直接記述せずに、通常のオブジェクトを扱うようにデータベースを扱うことができるということです。
Pythonでは、ORMモジュールの1つに SQLAlchemy があります。
SQLAlchemy
SQLAlchemyは、Pythonでポピュラーに使用されているORMライブラリの1つのようです。
SQLite、Postgresql、MySQL、Oracleなどさまざまなエンジンにも対応しており、PythonでORM、DBの処理を行うならSQLAlchemyを第一候補で考えれば良さそうです。
今回は、Flaskを用いてAPIとしてDBの情報を扱いたいので、FlaskにSQLAlchemyの機能を追加したFlask SQLAlchemyを使用します。
インストール
FlaskでREST APIを作成するので、Flask-RESTfulと取得した情報を扱いやすいようにflask-marshmallow もインストールします。
$ pip install Flask, Flask-RESTful, Flask-SQLAlchemy, flask-marshmallow
DBの準備
Dockerでmaridbを立ち上げます。
下記の内容でファイルを準備し、起動させます。
docker-compose.yml
version: "3"
services:
db:
image: mariadb
restart: always
ports:
- 3306:3306
command: --port 3306
environment:
- MYSQL_ROOT_PASSWORD=hogehoge.morita
- MYSQL_DATABASE=flask_test
mariadbの起動
$ docker-compose up -d
Flask側でのDB設定
インスタンスを生成し、初期化を行う関数を定義します。
この関数は、app.pyから実行されます。
database.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
def init_db(app):
db.init_app(app)
続いて、初期データを予め投入したいので、関数を準備します。
lib/create_init.py
from models.user import User
from database import db
def create_init(app):
with app.app_context():
db.drop_all()
db.create_all()
user_create()
db.session.commit()
def user_create():
# テストデータ
from test.data.user import users
for i in users:
u = User(name=i["name"])
db.session.add(u)
テストデータは、test/data/user.pyの中に記述します。
test/data/user.py
users = [
{ "name" : "山田太郎"},
{ "name" : "上田二郎"},
{ "name" : "田中三郎"}
]
DBの接続情報はconfig.pyで定義します。
接続情報は環境変数から読み出すようにしていますので.envの中に記述します。
config.py
class DevConfig:
# SQLAlchemy
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{user}:{password}@{host}/{dbname}?charset=utf8'.format(**{
'user': os.getenv('DB_USER', 'root'),
'password': os.getenv('DB_PASSWORD', ''),
'host': os.getenv('DB_HOST', 'localhost'),
'dbname': os.getenv('DB_NAME')
})
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
Config = DevConfig
.env
DB_USER="root" DB_PASSWORD="hogehoge.morita" DB_HOST="localhost" DB_NAME="flask_test"
ORMとして扱うモデルを定義
続いて、ORMとして扱うモデルを定義します。先程、定義したdbが必要なので、database.pyからimportします。
また、Marshmallowを用いることで、出力フォーマットが簡単に指定ができますので、こちらも定義します。
REST API用のクラスUserapiもここで定義します。
models/user.py
from flask import jsonify, abort, request
from flask_restful import Resource
from datetime import datetime
from database import db
from app import ma
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
device_id = db.Column(db.String(255), nullable=False, default="hoge")
name = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
updated_at = db.Column(db.DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
class UserSchema(ma.Schema):
class Meta:
# Fields to expose
fields = ("id", "name", "created_at")
users_schema = UserSchema(many=True)
class Userapi(Resource):
def get(self):
"""
ユーザを１件取得
"""
id = request.args.get('id')
result = User.query.filter_by(id=id).all()
if len(result) != 0:
return jsonify({"users" : users_schema.dump(result)})
else:
abort(404)
def post(self):
"""
ユーザ登録
"""
#ユーザを追加
print(request.json)
users = request.json["users"]
for i in users:
u = User(name=i["name"])
db.session.add(u)
db.session.commit()
return '', 204
def put(self):
"""
ユーザ更新
"""
id = request.json["id"]
user = User.query.filter_by(id=id).first()
if user :
for key, val in request.json.items():
setattr(user, key, val)
db.session.commit()
else:
abort(404)
return '', 204
def delete(self):
"""
ユーザ削除
"""
id = request.args.get('id')
user = User.query.filter_by(id=id).first()
if user :
db.session.delete(user)
db.session.commit()
return '', 204
else:
abort(404)
app.pyを作成
では、定義したDBの初期化関数、データの投入する関数を実行し、Flaskでサーバを起動させます。
app.py
# coding: utf-8
from flask import Flask
from flask_restful import Api
from flask_marshmallow import Marshmallow
from database import init_db
from lib.create_init import create_init
from models.user import Userapi
app = Flask(__name__)
app.config.from_object('config.Config')
# DB init
init_db(app)
api = Api(app)
ma = Marshmallow(app)
create_init(app)
# API add
api.add_resource(Userapi, '/user')
if __name__ == "__main__":
app.run(debug=True)
$ flask run
動作確認
正しく動作しているかを確認します。
GETメソッドでユーザ情報1件取得してみます。
user_test.py
import requests
URL = "http://127.0.0.1:5000/"
# GET test
id=1
res = requests.get(URL+"user?id={}".format(id))
print(res.json())
すると、確かにidが1のユーザ情報が取得できていることが確認できます。
{'users': [{'created_at': '2021-10-21T09:45:32', 'id': 1, 'name': '山田太郎'}]}
最後に
Flask-SQLAlchemyを用いて簡易的ですが、ORMを行ってみました。SQL文を記述しなくても良いのはとても便利ですね。
一方で、内部処理がブラックボックスとなっているので、SQLのチューニングする際などにはあまり向いていないようです。その際は従来通りSQL文で記述したほうが良さそうです。
ただ、個人使用であれば、オブジェクトとしてDBを扱えるのはとても便利ですので、今後も使用していきたいと思います。
