FlaskによるRESTful API
Vue.jsとFlaskを使ったフルスタックWeb開発に関する4番目のポストへようこそ。この投稿の焦点は、PythonベースのFlask Webフレームワークを使用して、バックエンドのREST APIを構築することです。
この投稿のコードは、私のGitHubアカウントのFourthPostというブランチのレポにあります。
シリーズ内容
- VueJSの紹介と概要
- Vueルーターの操作
- Vuexによる状態管理
- FlaskによるRESTful API (あなたはここにいます)
- REST APIによるAJAXの統合
- JWT認証
- 仮想プライベートサーバへの展開
フラスコの簡単な説明
Flask は Python ベースのマイクロフレームワークで、小規模から中規模の Web アプリケーションを迅速にプロトタイプ化し、開発するためのものです。Flask は既に StackAbuse のこことここの投稿で取り上げられているので、Flask の基本的な部分や一般的な部分について詳しく説明することはないでしょう。その代わり、今回までの記事で取り上げた、フロントエンドにデータを供給するためのRESTful APIを構築することに焦点を当てた、より実用的なアプローチを取るつもりです。
バックエンドプロジェクトファイルの足場固め
まず、/backend ディレクトリで Python3 の仮想環境を作り、Flask と他の必要なライブラリをいくつかインストールします。
$ python -m venv venv
$ source venv/bin/activate
(venv) $ pip install Flask Flask-SQLAlchemy Flask-Migrate Flask-Script requests
Flask（というかPythonのエコシステム全体）がとても素晴らしいのは、PyPIで利用できる多くのよくできたパッケージがあることです。以下は、私がインストールしたライブラリとその使用目的について簡単に説明したものです。
- Flask: ウェブマイクロフレームワーク
- Flask-SQLAlchemy: SQLAlchemyベースのORMで、Flask特有の素晴らしいソースがパッケージされています。
- Flask-Migrate: データベースマイグレーションライブラリ
- Flask-Script: スクリプト。Flask-Script: コマンドラインから Flask アプリケーションを操作するための非常に便利なパッケージです。
- リクエスト: REST API をテストするために使う、ネットワークリクエストを行うための便利なパッケージです。
backendディレクトリに、manage.pyとappserver.pyという新しいファイルを作成します。また、/backend ディレクトリの中に新しいディレクトリを作成し、これが私の “surveyapi” Flask アプリケーションとなります。surveyapi ディレクトリの中に、 _init_↵.py, models.py, application.py, api.py というファイルを作成します。この結果、/backendから始まるディレクトリ構成は以下のようになります（venvディレクトリは省略）。
├── manage.py
├── appserver.py
└── surveyapi
├── __init__.py
├── api.py
├── application.py
├── config.py
└── models.py
以下は、各ファイルがどのような目的で使用されるかを簡単に説明したものです。
- manage.py: Flaskアプリケーションのインスタンスにアクセスし、様々なFlask-Scriptコマンドを実行します。
- appserver.py: surveyapi アプリケーションを起動するスクリプトです。
- surveyapi/: バックエンドの Flask アプリケーション
- _init_JPY: Surveyapi ディレクトリを有効な Python パッケージにします。
- api.py: JSON リクエストとレスポンスを消費、生成する REST API ルートエンドポイントを定義します。
- application.py: Flask アプリケーションのインスタンスを作成します。
- config.py: Flaskアプリケーションのコンフィギュレーション設定を含みます。
- models.py: Survey、Question、Choice などのアンケートアプリケーションのデータオブジェクトとなるクラスを定義します。
アプリケーションファクトリの作成
まず、config.py の中でいくつかの設定を行い、surveyapi アプリケーションのコーディングを開始します。
"""
config.py
- settings for the flask application object
"""
class BaseConfig(object):
DEBUG = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///survey.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# used for encryption and session management
SECRET_KEY = 'mysecretkey'
この設定クラスは、survey.db という単一ファイルの SQLite データベースへの SQLALCHEMY_DATABASE_URI アプリケーションデータベース接続 URI を定義しています。また、暗号化に使用する SECRET_KEY 設定オプションも提供します。
application.py の内部で、アプリケーションファクトリ関数と呼ばれるものを作成します。これは、その名の通り、Flask アプリケーションのインスタンスを作成します。Flaskのインスタンスを生成するだけでなく、BaseConfigオブジェクトをソースして、次に作成するAPIルートのブループリントを登録します。
"""
application.py
- creates a Flask app instance and registers the database object
"""
from flask import Flask
def create_app(app_name='SURVEY_API'):
app = Flask(app_name)
app.config.from_object('surveyapi.config.BaseConfig')
from surveyapi.api import api
app.register_blueprint(api, url_prefix="/api")
return app
ブループリント API
次に、RESTful ルートを含む api という Blueprint オブジェクトを定義できる api.py モジュールに移動します。シンプルにするために、エンドポイント /api/hello/<string:name>/ に関連する say_hello() というシンプルなビュー関数を定義することから始めてみます。URL の <string:name> の部分は動的な文字列変数で、ビュー関数 say_hello(name) に関数パラメータとして渡され、返される JSON 応答メッセージで使用されます。
"""
api.py
- provides the API endpoints for consuming and producing
REST requests and responses
"""
from flask import Blueprint, jsonify, request
api = Blueprint('api', __name__)
@api.route('/hello/<string:name>/')
def say_hello(name):
response = { 'msg': "Hello {}".format(name) }
return jsonify(response)
Devサーバーのエントリーポイントとセットアップの検証
これをテストするために、appserver.py に数行のコードを追加して、アプリのインスタンスを作成する必要があります。これにより、appインスタンスの run() メソッドを呼び出すことで、Flask開発サーバーを起動することができます。
"""
appserver.py
- creates an application instance and runs the dev server
"""
if __name__ == '__main__':
from surveyapi.application import create_app
app = create_app()
app.run()
Flask開発サーバーを起動するには、Pythonインタプリタを起動し、以下のようにappserver.pyスクリプトを実行します。
(venv) $ python appserver.py
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 676-284-544
新しいエンドポイントをテストするために、仮想環境を有効にした新しいターミナルで Python インタープリターを起動し、 requests パッケージを使用して http://localhost:5000/api/hello/adam/ に GET リクエストを実行します。
(venv) $ python
>>> import requests
>>> response = requests.get('http://localhost:5000/api/hello/adam/')
>>> print(response.json())
{'msg': 'Hello adam'}
データレイヤーの定義
さて、Flask アプリケーションが機能していることが確認できたので、Flask-SQLAlchemy ORM の助けを借りて、データレイヤーを構築することに集中します。データレイヤーを実装するためには、models.pyの中に以下のようなデータクラスを書く必要があります。
- Survey: これはトップレベルのオブジェクトで、1つまたは複数の質問とその選択肢を含みます。
- 質問: アンケートオブジェクトに属し、選択肢を含むオブジェクト
- Choice: 質問に属するオブジェクトで、アンケートの質問に対する選択肢を表します。
これらのデータクラスは、Vue.jsのフロントエンドアプリケーションの構築に関する記事で以前に説明したものを模倣したフィールドを提起しますが、これらはデータが永続化されるデータベーステーブルにマッピングされることになります。
"""
models.py
- Data classes for the surveyapi application
"""
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Survey(db.Model):
__tablename__ = 'surveys'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
questions = db.relationship('Question', backref="survey", lazy=False)
def to_dict(self):
return dict(id=self.id,
name=self.name,
created_at=self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
questions=[question.to_dict() for question in self.questions])
class Question(db.Model):
__tablename__ = 'questions'
id = db.Column(db.Integer, primary_key=True)
text = db.Column(db.String(500), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
survey_id = db.Column(db.Integer, db.ForeignKey('surveys.id'))
choices = db.relationship('Choice', backref='question', lazy=False)
def to_dict(self):
return dict(id=self.id,
text=self.text,
created_at=self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
survey_id=self.survey_id,
choices=[choice.to_dict() for choice in self.choices])
class Choice(db.Model):
__tablename__ = 'choices'
id = db.Column(db.Integer, primary_key=True)
text = db.Column(db.String(100), nullable=False)
selected = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
question_id = db.Column(db.Integer, db.ForeignKey('questions.id'))
def to_dict(self):
return dict(id=self.id,
text=self.text,
created_at=self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
question_id=self.question_id)
前述したように、このアプリケーションの ORM には Flask-SQLAlchemy という SQLAlchemy の Flask 固有の拡張を使っています。Flask-SQLAlchemy が好きな理由は、かなり Pythonic な API を持っていて、データクラスを定義したり操作したりするのに賢明なデフォルトを提供してくれるからです。
各クラスは SQLAlchemy の Model ベースクラスを継承しており、データベースに保存されているデータを操作するための、直感的で読みやすいユーティリティメソッドを提供します。さらに、各クラスは、SQLAlchemy の Column クラスと関連する型 (Integer, String, DateTime, Text, …) で指定された、データベースのテーブルフィールドに変換される一連のクラスフィールドから構成されています。
また、各クラスは共通の to_dict() メソッドを持っていることにも気がつくでしょう。このメソッドは、フロントエンドクライアントにデータを送信する際に、モデルのデータをJSONにシリアライズするのに便利でしょう。
次の作業は、application.py で SQLAlchemy のオブジェクトである db を Flask のアプリケーションオブジェクトに登録することです。
"""
application.py
- creates a Flask app instance and registers the database object
"""
from flask import Flask
def create_app(app_name='SURVEY_API'):
app = Flask(app_name)
app.config.from_object('surveyapi.config.BaseConfig')
from surveyapi.api import api
app.register_blueprint(api, url_prefix="/api")
from surveyapi.models import db
db.init_app(app)
return app
最後に、Flask-Script と Flask-Migrate 拡張パッケージを manage.py モジュールの中にまとめて、マイグレーションを可能にすることです。この便利なモジュール manage.py は、先ほど定義したデータクラスをまとめ、Flask-Migrate と Flask-Script の仕組みと一緒にアプリケーションコンテキストにリンクしてくれます。
"""
manage.py
- provides a command line utility for interacting with the
application to perform interactive debugging and setup
"""
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from surveyapi.application import create_app
from surveyapi.models import db, Survey, Question, Choice
app = create_app()
migrate = Migrate(app, db)
manager = Manager(app)
# provide a migration utility command
manager.add_command('db', MigrateCommand)
# enable python shell with application context
@manager.shell
def shell_ctx():
return dict(app=app,
db=db,
Survey=Survey,
Question=Question,
Choice=Choice)
if __name__ == '__main__':
manager.run()
上記のコードで2つのことを行っています。まず、Flaskアプリケーションオブジェクトのインスタンスを作成して、 Migrate(app, db) と Manage(app) インスタンスにコンテキストを提供する。それから、managerオブジェクトにコマンドを追加して、コマンドラインからMigrationを作成・実行できるようにしています。
(venv) $ python manage.py db init
- migrations ディレクトリを surveyapi アプリケーションとデータベースファイル survey.db の隣に初期化します。
(venv) $ python manage.py db migrate
- models.py のクラスを SQL に変換し、対応するテーブルを生成するための初期マイグレーションファイルを作成します。
(venv) $ python manage.py db upgrade
- マイグレーションを実行し、前のステップで説明したテーブルでデータベースをアップグレードします。
この関数は、キーワードを app と db オブジェクトにマッピングして、 Survey, Question と Choice データクラスとともにディクショナリーを返します。
このシェルユーティリティコマンドの便利さを利用して、Flask-SQLAlchemy ORM を、このコマンドが生成する Python インタープリターで操作する方法を紹介します。
(venv) $ python manage.py shell
(venv) Adams-MacBook-Pro:backend adammcquistan$ python manage.py shell
>>> survey = Survey(name='Dogs')
>>> question = Question(text='What is your favorite dog?')
>>> question.choices = [Choice(text='Beagle'), Choice(text='Rottweiler'), Choice(text='Labrador')]
>>> question2 = Question(text='What is your second favorite dog?')
>>> question2.choices = [Choice(text='Beagle'), Choice(text='Rottweiler'), Choice(text='Labrador')]
>>> survey.questions = [question, question2]
>>> db.session.add(survey)
>>> db.session.commit()
>>> surveys = Survey.query.all()
>>> for s in surveys:
... print('Survey(id={}, name={})'.format(s.id, s.name))
... for q in s.questions:
... print(' Question(id={}, text={})'.format(q.id, q.text))
... for c in q.choices:
... print(' Choice(id={}, text={})'.format(c.id, c.text))
...
Survey(id=1, name=Dogs)
Question(id=1, text=What is your favorite dog?)
Choice(id=1, text=Beagle)
Choice(id=3, text=Labrador)
Choice(id=2, text=Rottweiler)
Question(id=2, text=What is your second favorite dog?)
Choice(id=4, text=Beagle)
Choice(id=6, text=Labrador)
Choice(id=5, text=Rottweiler)
これはかなりスマートでしょう？
ORMのエレガントで読みやすい構文についてだけでなく、アプリケーションのコンテキストを含むPythonインタプリタを起動して、アプリケーションのモデルでちょっとした実験を素早く行うことができるという、信じられないほど強力な能力について話しています。私は、バックエンドアプリケーションを構築するときに、この機能によってどれほど生産性が向上したかわかりませんし、同じことをするときに、この機能を利用することを真剣にお勧めします。
RESTful APIの完成
データアクセスレイヤーが構築されたので、RESTful API に必要な実装を完成させることに集中することができます。これは、アンケート、質問、選択肢のデータなどのアプリケーションリソースの消費と返却を処理するものです。RESTful API に必要なユースケースは以下のとおりです。
- すべてのアンケートとその質問および選択肢を取得する
- 単一のアンケートとその質問および選択肢をフェッチする
- 指定された質問と選択肢を含む新しいアンケートを作成する
- アンケートが実施された後に、アンケートの回答の選択肢を更新する
まず始めに、全てのデータクラスとSQLAlchemyの db インスタンスをインポートし、アクセスできるようにします。api.pyの先頭で、以下のインポートを追加します。
"""
api.py
- provides the API endpoints for consuming and producing
REST requests and responses
"""
from flask import Blueprint, jsonify, request
from .models import db, Survey, Question, Choice
実際のリソースエンドポイントについては、まずすべての調査リソースを取得する機能をコーディングします。api.py内で、 /hello/<string:name>/ のエンドポイントを、ルート /surveys/ のエンドポイントと surveys() ビュー関数に置き換える必要があります。
@api.route('/surveys/')
def surveys():
surveys = Survey.query.all()
return jsonify({ 'surveys': [s.to_dict() for s in surveys] })
もし開発サーバーがまだ動いているなら、プロジェクトファイルを保存すると、サーバーは自動的にすべての変更をリフレッシュして再読み込みするはずです。そうでなければ、 (venv) $ python appserver.py を実行すると、サーバーが起動します。これで、仮想環境を有効にした別のターミナルで requests パッケージを使用して、この新しいエンドポイントをテストすることができます。しかし、pprintと呼ばれる別の素晴らしいPythonパッケージを使って、JSONレスポンスをより読みやすい方法で表示するためのプロヒントを共有したいと思います。
(venv) $ pip install pprint
(venv) $ python
>>> import pprint, requests
>>> pp == pprint.PrettyPrinter()
>>> resp = requests.get('http://localhost:5000/api/surveys/')
>>> pp.pprint(resp.json())
{'surveys': [{
'created_at': '2018-03-06 03:52:44',
'id': 1,
'name': 'Dogs',
'questions': [{
'choices': [{
'created_at': '2018-03-06 03:52:44',
'id': 1,
'question_id': 1,
'text': 'Beagle'
},{
'created_at': '2018-03-06 03:52:44',
'id': 3,
'question_id': 1,
'text': 'Labrador'
},{
'created_at': '2018-03-06 03:52:44',
'id': 2,
'question_id': 1,
'text': 'Rottweiler'}],
'created_at': '2018-03-06 03:52:44',
'id': 1,
'survey_id': 1,
'text': 'What is your favorite dog?'
},{
'choices': [{
'created_at': '2018-03-06 03:52:44',
'id': 4,
'question_id': 2,
'text': 'Beagle'
},{
'created_at': '2018-03-06 03:52:44',
'id': 6,
'question_id': 2,
'text': 'Labrador'
},{
'created_at': '2018-03-06 03:52:44',
'id': 5,
'question_id': 2,
'text': 'Rottweiler'}],
'created_at': '2018-03-06 03:52:44',
'id': 2,
'survey_id': 1,
'text': 'What is your second favorite dog?'}]}
]}
次に、URL エンドポイント /surveys/id/ とビュー関数 survey(id) を使って、id によって一つのアンケートを取得する機能を実装してみます。API ビュー関数 surveys() の直後には、以下のコードを配置します。
@api.route('/surveys/<int:id>/')
def survey(id):
survey = Survey.query.get(id)
return jsonify({ 'survey': survey.to_dict() })
もう一度、ファイルを保存して、新しい API エンドポイントをテストし、有効なレスポンスを提供することを確認します。
>>> resp = requests.get('http://localhost:5000/api/surveys/1/')
>>> pp.pprint(resp.json())
{'survey': {
'created_at': '2018-03-06 03:52:44',
'id': 1,
'name': 'Dogs',
'questions': [{
'choices': [{
'created_at': '2018-03-06 03:52:44',
'id': 1,
'question_id': 1,
'text': 'Beagle'
},{
'created_at': '2018-03-06 03:52:44',
'id': 3,
'question_id': 1,
'text': 'Labrador'
},{
'created_at': '2018-03-06 03:52:44',
'id': 2,
'question_id': 1,
'text': 'Rottweiler'}],
'created_at': '2018-03-06 03:52:44',
'id': 1,
'survey_id': 1,
'text': 'What is your favorite dog?'
},{
'choices': [{
'created_at': '2018-03-06 03:52:44',
'id': 4,
'question_id': 2,
'text': 'Beagle'
},{
'created_at': '2018-03-06 03:52:44',
'id': 6,
'question_id': 2,
'text': 'Labrador'
},{
'created_at': '2018-03-06 03:52:44',
'id': 5,
'question_id': 2,
'text': 'Rottweiler'}],
'created_at': '2018-03-06 03:52:44',
'id': 2,
'survey_id': 1,
'text': 'What is your second favorite dog?'}]}
}
ここまでは、RESTful API からデータを取得するのに適した、デフォルトの HTTP GET ルートメソッドのみを使用してきました。しかし、最後の2つの機能については、エンドポイント /api/surveys/ と /api/surveys/id/ に対して、それぞれ HTTP POST と PUT メソッドを使用する必要があります。新しいアンケートを作成するには HTTP POST メソッドを使用し、既存のアンケートを新しい回答選択肢のセットで更新するには HTTP PUT メソッドを使用することにします。
api/surveys/ルートでは、ルート宣言にメソッドパラメータを追加して、GET と POST の両方のメソッドを受け入れることを指定します（methods=(‘GET’,’POST’)）。さらに、surveys()` ビュー関数の本体を変更して、メソッドの種類を区別し、新しいアンケートをデータベースに保存する機能を追加する予定です。
@api.route('/surveys/', methods=('GET', 'POST'))
def fetch_surveys():
if request.method == 'GET':
surveys = Survey.query.all()
return jsonify({ 'surveys': [s.to_dict() for s in surveys] })
elif request.method == 'POST':
data = request.get_json()
survey = Survey(name=data['name'])
questions = []
for q in data['questions']:
question = Question(text=q['text'])
question.choices = [Choice(text=c['text'])
for c in q['choices']]
questions.append(question)
survey.questions = questions
db.session.add(survey)
db.session.commit()
return jsonify(survey.to_dict()), 201
再度、プロジェクトを保存して、完全に機能するアンケート保存リソースがあることを確認するために、これをテストしたいと思います。
>>> import json
>>> survey = {
... 'name': 'Cars',
... 'questions': [{
... 'text': 'What is your favorite car?',
... 'choices': [
... { 'text': 'Corvette' },
... { 'text': 'Mustang' },
... { 'text': 'Camaro' }]
... }, {
... 'text': 'What is your second favorite car?',
... 'choices': [
... { 'text': 'Corvette' },
... { 'text': 'Mustang' },
... { 'text': 'Camaro' }]
... }]
... }
>>> headers = {'Content-type': 'application/json'}
>>> resp = requests.post('http://localhost:5000/api/surveys/', headers=headers, data=json.dumps(survey))
>>> resp.status_code
201
最後に実装するのは、新しいアンケートの回答選択で既存のアンケートを更新する機能です。ここでも、GET と PUT のメソッドを /api/surveys/id/ のルート定義、methods=('GET', 'PUT') に追加する必要があります。それから、PUT リクエストの JSON 本文で選択されていると指定された、関連するアンケートの質問の選択肢を更新するために survey(id) ビュー関数を更新しています。
@api.route('/surveys/<int:id>/', methods=('GET', 'PUT'))
def survey(id):
if request.method == 'GET':
survey = Survey.query.get(id)
return jsonify({ 'survey': survey.to_dict() })
elif request.method == 'PUT':
data = request.get_json()
for q in data['questions']:
choice = Choice.query.get(q['choice'])
choice.selected = choice.selected + 1
db.session.commit()
survey = Survey.query.get(data['id'])
return jsonify(survey.to_dict()), 201
最後に、すべてのファイルを保存して、このように最終テストを行う必要があります。
>>> survey_choices = {
... 'id': 1,
... 'name': 'Dogs',
... 'questions': [
... { 'id': 1, 'choice': 1 },
... { 'id': 2, 'choice': 5 }]
... }
>>> headers = {'Content-type': 'application/json'}
>>> resp = requests.put('http://localhost:5000/api/surveys/1/', data=json.dumps(survey_choices), headers=headers)
>>> resp.status_code()
201
リソース
PythonとバックエンドAPI構築についてもっと学びたいですか？REST APIs with Flask and Pythonのようなコースをチェックして、Pythonを使ったバックエンドのWeb開発について深く掘り下げてみてください。
結論
今回は、Flaskを使ったシンプルなRESTful APIを以下の表に従って実装する方法を取り上げました。
| ルート│メソッド│機能││││。
| — | — | — |
/api/surveys/ | GET | すべてのアンケートを取得する。
/api/surveys/ | POST | 新しいアンケートを作成する。
| /api/surveys/id/ | GET | アンケートを ID で取得する
| /api/surveys/id/ | PUT | アンケートの選択項目を更新する｜(英語)
次回は、フロントエンドの Vue.js アプリケーションを統合して、データの更新を Flask バックエンドにプッシュできるようにする方法を紹介する予定です。
また、コメントや批評をお待ちしています。
</strong