FlaskとSQLAlchemyによるWebサービスの実装
"FlaskによるWebサービスの実装 Part2"では、実際の運用にはデータベースのサポートが必要と指摘しました。
Flask-SQLAlchemy を使ってデータベースをサポートしたWebサービスTODOを実装するようにしましょう。
HTTPメソッドのマッピング
WebサービスTODOの仕様を少し変更しています。
本来であればユーザリソースについてAPIで操作できるようにするべきですが、今回は理解が容易になるようタスクリソースだけとしておきます。
また、単純化するための認証は省いています。
table: APIとHTTPメソッド
HTTPメソッド URI アクション
GET http://hostname/todo/api/v4.0/tasks タスクリストの取得
GET http://hostname/todo/api/v4.0/tasks/id タスクの取得
POST http://hostname/todo/api/v4.0/tasks 新規タスクの作成
PUT http://hostname/todo/api/v4.0/tasks/id 既存タスクを更新
DELETE http://hostname/todo/api/v4.0/tasks/id 指定したタスクを削除
タスクリソースは次の情報を持つものとします。
uri：タスクを示す一意のURI。String型。
title：タスクのタイトル。タスクについての短い説明。 String型。
description：タスクの詳細。タスクについての詳細な説明。 Text型。
done：タスクの完了状態。 Boolean型。
Marshmallow は複雑なオブジェクトをシリアライズ/ディシリアライズするためのライブラリで、SQLAlchemy と共に使うとモデルクラスからJSON形式に簡単に変換することができるようになります。Flask-RESTful の ヘルパー関数marshal()と似ていますが、Marshamallow はSQLAlchemy のモデルクラスを理解してくれるためフィールド定義が不要になります。
インストール
code: bash
$ pip install -U flask-sqlalchemy marshmallow-sqlalchemy flask-marshmallow
準備
アプリケーションのディレクトリを作成します。
code: bash
$ mkdir -p $HOME/flask/todo_apiv4
$ cd $HOME/flask/todo_apiv4
はじめに、 次のapp.py を用意します。
SQLAlchemy は Mashmallow の前に初期化を行うようにします。
code: config.py
import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config(object):
SECRET_KEY = os.environ.get('SECRET_KEY') or 'super-secret-key'
DEBUG = True
CSRF_ENABLED = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'todo.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
初期化のための app.py を作成します。
code: python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_marshmallow import Marshmallow
from flask_restful import Api, Resource, reqparse
from config import Config
app = Flask(__name__)
app.config.from_object(Config)
db = SQLAlchemy(app)
ma = Marshmallow(app)
migrate = Migrate(app, db)
@app.shell_context_processor
def make_shell_context():
return {'db': db, 'Task': Task, 'TaskSchema': TaskSchema }
if __name__ == '__main__':
app.run(port=8080)
make_shell_context() は flask shell を実行したときに意味があるものなので、省略してもかまいません。
モデルクラスの定義
データベースで作成する必要があるものは、タスクのためのTaskテーブルです。
モデルクラスとスキーマクラスを定義しましょう。
スキーマクラスはMarshmallowがモデルクラスとJSONとの変換のために使用するものです。
code: python
class Task(db.Model):
__tablename__ = 'tasks'
id = db.Column(db.Integer, primary_key = True)
title = db.Column(db.String(128), index = True)
discription = db.Column(db.Text(256))
done = db.Column(db.Boolean)
def __init__(self, title, description, done=False):
self.title = title
self.description = description
self.done = done
def __repr__(self):
return f'Task: {self.title}'
class TaskSchema(ma.SQLAlchemySchema):
class Meta:
model = Task
uri = ma.URLFor("task", id='<id>')
task_schema = TaskSchema()
tasks_schema = TaskSchema(many=True)
URLFor() はエンドポイントのURLを出力するフィールドタイプです。 オブジェクトから引数を取得してシリアル化できることを除いて、Flaskのヘルパー関数url_for()と同じように機能します。
APIの定義
Flask-RESTful を使ってリクエストデータの解析のためのリソースクラスを定義します。
これは、"Flaskの拡張機能を使ったRESTful APIの実装" で説明したものと変わりません。
code: python
class TaskListAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title',
type = str, required = True,
help = 'No task title provided',
location = 'json')
self.reqparse.add_argument('description',
type = str, default = "",
location = 'json')
super(TaskListAPI, self).__init__()
def get(self):
pass
def post(self):
pass
class TaskAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title',
type = str, location = 'json')
self.reqparse.add_argument('description',
type = str, location = 'json')
self.reqparse.add_argument('done',
type = bool, location = 'json')
super(TaskAPI, self).__init__()
def get(self, id):
pass
def put(self, id):
pass
def delete(self, id):
pass
api.add_resource(TaskListAPI,
'/todo/api/v4.0/tasks', endpoint = 'tasks')
api.add_resource(TaskAPI,
'/todo/api/v4.0/tasks/<int:id>', endpoint = 'task')
TaskListAPIリソースクラスのメソッドを定義
まず get()メソッドを見てみます。RESTful でHTTPプロトコルのGETメソッドで呼び出される関数です。"Flaskの拡張機能を使ったRESTful APIの実装" と違う点はデータベースにアクセスする必要があるということです。
code: python
class TaskListAPI(Resource):
# ...
def get(self):
tasks = Task.query.all()
return tasks_schema.dump(tasks)
def post(self):
args = self.reqparse.parse_args()
task = Task(
title = args'title',
description = args'description',
)
db.session.add(task)
db.session.commit()
return task_schema.dump(task)
TaskAPIリソースクラスのメソッドを定義
TaskListAPIリソースクラスとほぼ同じです。
code: python
class TaskAPI(Resource):
# ...
def get(self, id):
post = Task.query.get_or_404(id)
return task_schema.dump(post)
def put(self, id):
args = self.reqparse.parse_args()
task = Task.query.get_or_404(id)
task.title = args'title'
task.description = args'description'
task.done = args'done'
db.session.commit()
return task_schema.dump(task)
def delete(self, id):
task = Task.query.get_or_404(id)
db.session.delete(task)
db.session.commit()
return '', 204
データベースへのアクセスは SQLAlchemy ORM を介して処理されます。
応答の生成には mashmallow のスキーマクラスを使ってオブジェクトのJSONに変換します。
ここで、注目すべきところは、Flask-RESTful のmarshal()に変えて、mashmallow で定義したスキーマクラスによって dump() することでJSON出力がされるということです。
データベースの作成
Flask-Migrate を設定しているのでデータベースの作成は次の手順で行います。
code: bash
$ flask db init
$ flask db migrate -m "db initialized"
$ flask db upgrade
ここまでのコードの整理
code: app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_marshmallow import Marshmallow
from flask_restful import Api, Resource, reqparse, marshal
from config import Config
app = Flask(__name__)
app.config.from_object(Config)
db = SQLAlchemy(app)
ma = Marshmallow(app)
migrate = Migrate(app, db)
api = Api(app)
class Task(db.Model):
__tablename__ = 'tasks'
id = db.Column(db.Integer, primary_key = True)
title = db.Column(db.String(128), index = True)
description = db.Column(db.Text(256))
done = db.Column(db.Boolean)
def __init__(self, title, description, done=False):
self.title = title
self.description = description
self.done = done
def __repr__(self):
return f'Task: {self.title}'
class TaskSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Task
uri = ma.URLFor("task", id='<id>')
task_schema = TaskSchema()
tasks_schema = TaskSchema(many=True)
class TaskListAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title',
type = str, required = True,
help = 'No task title provided',
location = 'json')
self.reqparse.add_argument('description',
type = str, default = "",
location = 'json')
super(TaskListAPI, self).__init__()
def get(self):
tasks = Task.query.all()
return tasks_schema.dump(tasks)
def post(self):
args = self.reqparse.parse_args()
task = Task(
title = args'title',
description = args'description',
)
db.session.add(task)
db.session.commit()
return task_schema.dump(task)
class TaskAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title',
type = str, location = 'json')
self.reqparse.add_argument('description',
type = str, location = 'json')
self.reqparse.add_argument('done',
type = bool, location = 'json')
super(TaskAPI, self).__init__()
def get(self, id):
post = Task.query.get_or_404(id)
return task_schema.dump(post)
def put(self, id):
args = self.reqparse.parse_args()
task = Task.query.get_or_404(id)
task.title = args'title'
task.description = args'description'
task.done = args'done'
db.session.commit()
return task_schema.dump(task)
def delete(self, id):
task = Task.query.get_or_404(id)
db.session.delete(task)
db.session.commit()
return '', 204
api.add_resource(TaskListAPI,
'/todo/api/v4.0/tasks', endpoint = 'tasks')
api.add_resource(TaskAPI,
'/todo/api/v4.0/tasks/<int:id>', endpoint = 'task')
@app.shell_context_processor
def make_shell_context():
return {'db': db, 'Task': Task, 'TaskSchema': TaskSchema }
if __name__ == '__main__':
app.run(port=8080)
動作確認
はじめに Flask を起動します。
code: bash
$ python app.py
以下は、別のターミナルから実行します。
タスク一覧を取得
データベースにはまだ何も登録されていないのでからです。
code: bash
$ curl -i -X GET http://localhost:8080/todo/api/v4.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 3
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:42:20 GMT
[]
タスクの登録
code: bash
$ curl -i -H "Content-Type: application/json" -X POST -d '{"title":"Buy Beers", "description": "IPA 6 bottles"}' http://localhost:8080/todo/api/v4.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:47:33 GMT
{
"id": 1,
"done": false,
"description": "IPA 6 bottles",
"title": "Buy Beers",
"uri": "/todo/api/v4.0/tasks/1"
}
タスク一覧を再取得
code: bash
$ curl -i -X GET http://localhost:8080/todo/api/v4.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 166
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:49:10 GMT
[
{
"id": 1,
"done": false,
"description": "IPA 6 bottles",
"title": "Buy Beers",
"uri": "/todo/api/v4.0/tasks/1"
}
]
タスクの内容を更新
code: bash
$ curl -i -H "Content-Type: application/json" -X PUT -d '{"title":"Gymnastics", "description": "Goto Anytime Fitness"}' http://localhost:8080/todo/api/v4.0/tasks/1
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 141
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:52:19 GMT
{
"id": 1,
"done": null,
"description": "Goto Anytime Fitness",
"title": "Gymnastics",
"uri": "/todo/api/v4.0/tasks/1"
}
タスクを指定して取得
code: bash
$ curl -i -X GET http://localhost:8080/todo/api/v4.0/tasks/1
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 141
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:53:19 GMT
{
"id": 1,
"done": null,
"description": "Goto Anytime Fitness",
"title": "Gymnastics",
"uri": "/todo/api/v4.0/tasks/1"
}
タスクを削除
code: bash
$ curl -i -X DELETE http://localhost:8080/todo/api/v4.0/tasks/1
HTTP/1.0 204 NO CONTENT
Content-Type: application/json
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 06:53:47 GMT
タスクを指定して再取得
code: bash
$ curl -i -X GET http://localhost:8080/todo/api/v4.0/tasks/1
HTTP/1.0 404 NOT FOUND
Content-Type: application/json
Content-Length: 142
Server: Werkzeug/1.0.1 Python/3.6.10
Date: Fri, 15 May 2020 11:54:24 GMT
{
"message": "The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again."
}
まとめ
データベースと連携した RESTful Webサービスでは、以下の拡張機能を使うと簡単に実装できることが確認できました。
Flask-RESTful
Flask-SQLAlchemy
Flask-Migrate
Flask-Marshallow
Mashmallow-SQLAlchemy