はじめに
業務では「FuelPHP」という第２検索ワードで「オワコン」と出てくるフレームワークを使っているPHPエンジニアです。プログラミングを始めて、もうそろそろで２年が経とうとしています。（他の言語に手を出したくなる年頃です）
掲示板をSPA（シングルページアプリケーション）で作ったので、"SPA開発における"と銘打っていますが、記事の内容は「FlaskでREST APIを実装する」ということに終始しています。
FlaskでAPI開発するにあたり、いろいろな記事を参考にさせていただいたのですが、バリデーションや認証など一通りまとまっている日本語の記事がなかったので（よく知らないのですが、"小さく"始めるのがFlaskの思想？）、勉強したことをまとめてみたいと思います。
自分用のメモでもあるので、膨大な量になってしまっていますが、適宜気になるところだけでも読んでいただけたら幸いです。
REST APIについての説明は今回省きますが、下記の記事が参考になります。
また、API設計書をStoplight Studioというツールで作成しました。初めてAPI設計書のツールを使ったのですが、ツールに沿って作るだけでいい感じにRESTになるので設計がスムーズにできました。
参考：
また、Stoplight Platformを使えば、チーム内でAPI設計書を共有できます。
この記事の主な内容
-
SQLAlchemyとmarshmallowでORM
-
marshmallowでリクエストのバリデーション
-
Flask-RESTfulとBlueprintでエンドポイント作成
-
Flask-JWT-ExtendedでJWT認証
Flaskアプリ
まずは、Flaskをインストールします。
$ pip install Flask
ディレクトリ構成は以下のようにしました。
backend/
├── api.py # Flaskアプリの実行ファイル
├── config.py # Flaskアプリの設定ファイル
├── api/
│ ├── models/ # DBのテーブルごとにファイル作成
│ ├── validators/ # バリデーションが必要なリソースごとにファイル作成
│ ├── views/ # エンドポイントのリソースごとにファイル作成
│ ├── __init__.py
│ ├── database.py # FlaskアプリでDBを使うときの設定ファイル
│ ├── error.py # エラーレスポンスをまとめたファイル
│ └── token.py # Flask-JWT-Extendedの拡張ファイル
│
├── services/ # Flaskアプリに依存しない自作クラス（AWSなど）
├── migrations/ # Flask-Migrateで作成
└── requirements.txt # 必要なライブラリ（pip freezeの出力内容）
Flaskアプリにおいて基本となるファイルは、以下になります。
from api import app as application
if __name__ == '__main__':
application.run()
$ python api.pyを実行すると、ローカル環境でFlaskアプリが起動します。api/配下（api/__init__.py）にあるFlaskインスタンスを呼んで、アプリを起動します。
from flask import Flask
from .database import init_db # （２）セクション：DB操作 のときに追加する
from .token import init_jwt # （４）セクション：認証 のときに追加する
from .models import User # migrateを実行するときのために、モデル定義を読み込む（２）
from .views import api_bp # （３）セクション：エンドポイントとルーティング のときに追加する
app = Flask(__name__)
# Flaskアプリの設定（config.pyの内容）を読み込む
app.config.from_object('config.Config')
# DB設定を読み込む（２）
init_db(app)
# JWT認証の設定を読み込む（４）
init_jwt(app)
# エンドポイントのルーティング設定をFlaskアプリに適用（３）
app.register_blueprint(api_bp)
ここでFlaskアプリの設定・必要な情報の読み込みを行っています。Flaskインスタンスであるappをapi.py以外のファイルから呼ばないように意識しました。
一応、最終形はこんな感じでスッキリしています。
class DevelopmentConfig:
# Flask
ENV = 'development'
DEBUG = True
# JSONのレスポンスボディで日本語が文字化けしないために
JSON_AS_ASCII = False
'''SQLAlchemy（DB設定）（２）'''
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{user}:{password}@{host}/{db_name}?charset=utf8mb4'.format(**{
'user': 'xxxxx',
'password': 'xxxxx',
'host': 'xxxxx',
'db_name': 'xxxxx'
})
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
'''Flask-JWT-Extended（JWT認証）（３）'''
JWT_TOKEN_LOCATION = ['cookies']
# クッキーの有効期限を無効化する
JWT_ACCESS_TOKEN_EXPIRES = False
JWT_REFRESH_TOKEN_EXPIRES = False
# httpsでなくてもクッキーを送る
JWT_COOKIE_SECURE = False
# クッキーのpath属性
JWT_ACCESS_COOKIE_PATH = '/api/'
JWT_REFRESH_COOKIE_PATH = '/api/auth/refresh'
# GET以外のときにCSRFトークンチェックを行うか
JWT_COOKIE_CSRF_PROTECT = True
# JWT署名鍵
JWT_SECRET_KEY = b'xxxxx'
Config = DevelopmentConfig
こちらがFlaskアプリの設定ファイルです。こちらは開発環境用なので、デバッグモードをONにしていたり、認証トークンの有効期限を無期限にしたりしています。
以上がFlaskアプリの導入ですが、まだ登場していない内容も多いので、適宜読み飛ばしてください。
DB操作
DB関連で必要となるライブラリをインストールします。
$ pip install Flask-SQLAlchemy, Flask-Migrate, flask-marshmallow, marshmallow-sqlalchemy
それぞれを簡単に説明すると、
-
SQLAlchemy ・・・DB接続、ORM、SQLの実行
-
Migrate ・・・DBのテーブル定義、マイグレーション（今回は説明を省きます。こちらの記事を参考にさせていただきました→Flask + SQLAlchemyプロジェクトを始める手順）
-
marshmallow ・・・SQL実行で取得したモデルのオブジェクトをJSONで扱える形に変換する＆データ内容の整形（あとで、リクエストボディのバリデーションにも使います）
まず、api/database.pyを作成して、api/__init__.pyで呼ぶinit_db()を定義します。
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow, fields
from flask_migrate import Migrate
from sqlalchemy.dialects.mysql import TINYINT, VARBINARY
db = SQLAlchemy()
ma = Marshmallow()
fields = fields.fields
# flask_sqlalchemyに用意されていないデータ型をセット
db.Tinyint = TINYINT
db.Varbinary = VARBINARY
# 実行されるSQL文を出力させる（NOTE: 開発用）
# import logging
# logging.basicConfig()
# logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
def init_db(app):
db.init_app(app)
Migrate(app, db)
こちらのファイルは、モデルの定義やSQL実行の時に必要なので、今後いろいろなファイルから呼ばれます。
次に、モデルを作成します。こちらは、DBのテーブルに対応します。
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from ..database import db, ma
class User(db.Model):
# テーブル名を定義
__tablename__ = 'users'
# テーブルのカラムを定義
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(191), nullable=False)
email = db.Column(db.String(191), unique=True, nullable=False)
password = db.Column(db.String(191), nullable=False)
is_admin = db.Column(db.Tinyint(1), nullable=False, default=0)
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)
def __init__(self, name, email, password, is_admin):
self.name = name
self.email = email
self.set_password(password)
self.is_admin = is_admin
# パスワードをDBにハッシュ化して保存する
def set_password(self, password):
self.password = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password, password)
class UserSchema(ma.SQLAlchemyAutoSchema):
class Meta:
# モデルのプロパティ（テーブルのカラム）を全てスキーマに適用する
model = User
# dumpするときに表示させない
password = ma.auto_field(load_only=True)
このようにモデルを定義した後に、Flask-Migrateを使用すると、モデルの内容がテーブルとしてDBに反映されます。
モデルを作成してテーブルまでできると、SQLを実行できます。
from database import db
from models.user import User, UserSchema
"""Insert"""
# 適当にデータを用意（実際はリクエストボディ）
input_data = {
'name': 'user',
'email': 'user@example.com',
'password': 'userpass',
'is_admin': 1
}
new_user = User(**input_data)
# データベースに追加
db.session.add(new_user)
db.session.commit()
# new_userのままだとJSONに変換できないので、先ほど定義したUserSchema()で変換可能にする
# また、passwordを出力しないようにもなる
res = UserSchema().dump(new_user)
"""Select"""
# 全ユーザ取得
users = db.session.query(User).all()
# many=Trueで複数のレコードに対して、スキーマを適用できる
res = UserSchema().dump(users, many=True)
"""Update"""
# まず、変更対象のユーザを取得
user = db.session.query(User).filter_by(email='user@example.com').first()
# userはUserモデルのオブジェクトなので、プロパティを変更できる
user.name = 'superuser'
# 変更内容をDBにコミット
db.session.commit()
"""Delete"""
# データベースから物理削除
db.session.delete(user)
db.session.commit()
SQLAlchemyの使い方はこちらの記事が参考になりました。今回のようにFlask-SQLAlchemyで使う場合、db.を付けるように読み替えていけば適用できます。
→ pythonのORM：SQLAlchemy の基本的な使い方
テーブルの数だけモデルが増えて、importが複雑になるので、以下のように__init__.pyを用意して、モデルとスキーマをまとめます。
from .user import User, UserSchema
from .tag import Tag, TagSchema
# 省略
__all__ = [
User,
UserSchema,
Tag,
TagSchema,
# 省略
]
これでfrom ..models import User, UserSchema, Tagのようにして、他ファイルから1行ですべてのモデル・スキーマにアクセスできるようになります。
以上がDB操作に関してですが、marshmallowを使ってモデル用のスキーマを作成することで、Viewファイル（MVCでいうところのコントローラ）に書く処理を減らせます。また、各モデルの処理が対照的なので、慣れるとコードの見通しが良くなります。さらに、あとで紹介するDBのリレーション時に特に威力を発揮します。
バリデーション
リクエストボディ（パラメタ）のバリデーションをmarshmallowというライブラリを使って行います。先ほど、flask-marshmallowをインストールしたときに、marshmallowもインストールされるので、今回は特にインストールする必要はありません。
まず、バリデーション用のスキーマを作成します。こちらは、先ほどのモデルのスキーマとは全くの別物なので、注意してください。
from marshmallow import Schema, fields, validate
# GETリクエスト用のスキーマを定義
class GetUserListSchema(Schema):
# 0 or 1のみ許容
has_posted = fields.Int(validate=validate.OneOf([0, 1]))
# 50文字以下
keyword = fields.Str(validate=validate.Length(max=50))
# POSTリクエスト用のスキーマを定義
class PostUserListSchema(Schema):
# required=Trueで、キー（'name'）がなければエラー
name = fields.Str(required=True, validate=validate.Length(min=1, max=50))
# メールアドレスの形式になっているか
email = fields.Email(required=True)
# 正規表現でバリデーション。エラー内容も設定可能
password = fields.Str(required=True, validate=validate.Regexp(regex='^([a-zA-Z0-9]{8,50})$', error='8文字以上50文字以下の半角英数字で入力してください'))
is_admin = fields.Int(required=True, validate=validate.OneOf([0, 1]))
get_userlist_schema = GetUserListSchema()
post_userlist_schema = PostUserListSchema()
他にもPATCHリクエストのときに、プロフィール画像用のバリデーションを自作していたりしているのですが、今回は省きます。（バリデーションの自作については、こちらで確認できます）
細かい設定も公式で確認できます。
ここで作成したスキーマを使って、バリデーションを実行します。
from flask import request
from validators.user import get_userlist_schema
'''GETリクエスト'''
# クエリパラメータを辞書型として受け取る
input_data = request.args
# バリデーションを実行する
errors = get_userlist_schema.validate(input_data)
if errors:
# バリデーションエラーが発生した場合
return str(errors)
# 処理が続く
以上がバリデーションについてですが、上記のようにすることでViewファイルからバリデーションの処理を抽出できて、Viewファイルの記述を減らすことができます。
エンドポイントとルーティング
エンドポイントが多くても、比較的整理された状態でコーディングできるようにFlask-RESTfulをインストールします。（別にマストでもない気がします。ただ見通しが良くなります）
$ pip install Flask-RESTful
以下のようにして、リソースごとにViewファイルを作成して、エンドポイントの処理を記述します。
from flask import request
from flask_restful import Resource
from ..error import Abort
from ..database import db
from ..models import User, UserSchema, Article
from ..validators.user import get_userlist_schema, post_userlist_schema
'''エンドポイント：api/users'''
# HTTPメソッドごとに、メソッドを作成する
class UserListAPI(Resource):
'''ユーザ一覧取得'''
def get(self):
input_data = request.args
errors = get_userlist_schema.validate(input_data)
if errors:
Abort.error_4000(str(errors))
# パラメータがあれば、検索条件として追加していく
filters = []
if 'has_posted' in input_data and input_data['has_posted']=='1':
filters.append(Article.is_public==1)
if 'keyword' in input_data:
filters.append(User.name.like('%'+input_data['keyword']+'%'))
users = (db.session.query(User)
.outerjoin(Article, db.and_(Article.user_id==User.id, Article.is_public==1))
.filter(db.and_(*filters))
).all()
users_data = UserSchema().dump(users, many=True)
return jsonify({'users': users_data, 'hit_count': len(users_data)})
'''ユーザ新規作成'''
def post(self):
input_data = request.json
errors = post_userlist_schema.validate(input_data)
if errors:
Abort.error_4000(str(errors))
# 同じメールアドレスは登録できない
user = db.session.query(User).filter_by(email=input_data['email']).first()
if user:
Abort.error_4001('email: 既に登録されているメールアドレスです')
new_user = User(**input_data)
# データベースに追加
db.session.add(new_user)
db.session.commit()
res = UserSchema().dump(new_user)
return res, 201
'''エンドポイント：api/users/{id}'''
# HTTPメソッドごとに、メソッドを作成する
class UserAPI(Resource):
'''ユーザ取得'''
def get(self, id):
# 省略
'''ユーザ編集'''
def patch(self, id):
# 省略
'''ユーザ削除'''
def delete(self, id):
# 省略
続いて、以下のようにリソースごとに作成したエンドポイントをBlueprintに追加していき、最終的に出来上がったBlueprintインスタンスをapi/__init__.pyでregister_blueprint()をして、Flaskアプリのルーティング設定を行います。
from flask import Blueprint
from flask_restful import Api
from .user import UserListAPI, UserAPI
from .auth import AuthLoginAPI, AuthRefreshAPI, AuthLogoutAPI
# 省略
# api/__init__.pyで呼ぶためのBlueprintインスタンスを作成
api_bp = Blueprint('api', __name__, url_prefix='/api')
api = Api(api_bp)
# 先ほど定義したResourceクラス（エンドポイント）をURLマッピングしていく
api.add_resource(UserListAPI, '/users')
api.add_resource(UserAPI, '/users/<int:id>')
api.add_resource(AuthLoginAPI, '/auth/login')
api.add_resource(AuthRefreshAPI, '/auth/refresh')
api.add_resource(AuthLogoutAPI, '/auth/logout')
# 省略
このようにして、全エンドポイント（ルーティング）を１ファイルで定義することで、REST APIの全容がつかみやすくなります。
また、Viewファイルのところで出てきたAbortですが、以下のファイルで定義しています。
from flask_restful import abort
class Abort:
BAD_REQUEST_STATUS = 400
UNAUTHORIZED_STATUS = 401
FORBIDDEN_STATUS = 403
NOT_FOUND_STATUS = 404
INTERNAL_SERVER_ERROR_STATUS = 500
# リクエストパラメータ・ボディの型エラー
@classmethod
def error_4000(cls, message):
abort(cls.BAD_REQUEST_STATUS, message=message, code=4000)
# リクエストパラメータ・ボディのデータ不整合エラー
@classmethod
def error_4001(cls, message):
abort(cls.BAD_REQUEST_STATUS, message=message, code=4001)
# 省略
このような形でエラーコードごとにメソッドを作成してクラスにまとめることで、エラーレスポンスに統一感を持たせています。
認証
やっとSPA開発っぽい内容ですが、認証には「JWT認証(クッキー保存)+CSRFトークン」を採用しました。
SPAの認証については、こちらの記事が大変参考になりました。
→ SPAのログイン認証のベストプラクティスがわからなかったのでわりと網羅的に研究してみた〜JWT or Session どっち？〜
上の記事の中で、
CSRFトークン（レスポンスヘッダやレスポンスボディなどで返却。Set-Cookieでの返却だとCSRF対策にならない。）
を返却
とあったのですが、基本的にWebサイトで実行されているJavaScriptコードは他のWebサイトのCookieを読み取ることができないはずなので、CSRFトークンもCookieに入れることにしました。（下で紹介するFlask-JWT-Extendedのクッキー認証のデフォルトのまま）
このあたりはあまり自信がないので、間違えていたらご指摘よろしくお願いします。
まず、JWT認証(クッキー保存)を行うためのライブラリをインストールします。
$ pip install Flask-JWT-Extended
次に、api/token.pyを作成して、api/__init__.pyで呼ぶinit_jwt()を定義します。
from flask_jwt_extended import (
JWTManager, verify_jwt_in_request, get_jwt_claims, get_jwt_identity,
jwt_required, create_access_token,
jwt_refresh_token_required, create_refresh_token,
set_access_cookies, set_refresh_cookies, unset_jwt_cookies
)
from functools import wraps # デコレータ作成用
from .error import Abort
from .models import User, Article, Comment
# 自作のメソッドを適用するために、最初にインスタンスを用意
jwt = JWTManager()
'''以下で、JWTの挙動をデフォルトから変更する(エラーのフォーマットを変えるなど)'''
# アクセストークンに、権限情報も含めるようにする
@jwt.user_claims_loader
def add_claims_to_access_token(identity):
user = User.query.filter_by(id=identity).first()
return {
'id': identity,
'is_admin': user.is_admin
}
# トークンの有効期限切れ時の挙動
@jwt.expired_token_loader
def my_expired_token_callback(expired_token):
token_type = expired_token['type']
if token_type == 'access':
Abort.error_4011() # アクセストークンの有効期限切れ
else:
Abort.error_4012() # リフレッシュトークンの有効期限切れ
# 無効な形式のトークン時の挙動
@jwt.invalid_token_loader
def my_invalid_token_callback(error_string):
Abort.error_4013(error_string)
# 認証エラー時の挙動
@jwt.unauthorized_loader
def my_unauthorized_callback(error_string):
Abort.error_4010(error_string)
def init_jwt(app):
jwt.init_app(app)
'''アクセス制限用にデコレータを作成する'''
# 管理者のみアクセスできるエンドポイント
def admin_required(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
verify_jwt_in_request() # 通常のトークン認証
claims = get_jwt_claims()
if claims['is_admin'] != 1: # カスタマイズした権限情報の確認
Abort.error_4030()
else:
return fn(*args, **kwargs)
return wrapper
# 本人と管理者のみアクセスできるエンドポイント
def self_or_admin_required_user(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
# kwargs: 修飾する関数の引数（対象のユーザID）が入る
verify_jwt_in_request()
claims = get_jwt_claims()
if claims['id'] == kwargs['id'] or claims['is_admin'] == 1:
return fn(*args, **kwargs)
else:
Abort.error_4031()
return wrapper
Flask-JWT-Extendedのデフォルト挙動の変更については、公式に書いてあります。
以下のように、各エンドポイントのHTTPメソッドごとにアクセス制限を設定していきます。
from flask import request
from flask_restful import Resource
from ..error import Abort
from ..database import db
from ..token import jwt_required, admin_required, self_or_admin_required_user # 追加
from ..models import User, UserSchema
from ..validators.user import post_userlist_schema
'''エンドポイント：api/users'''
class UserListAPI(Resource):
'''ユーザ一覧取得'''
@jwt_required # 通常のトークン認証（ログイン中か）
def get(self):
# 省略
'''ユーザ新規作成'''
@admin_required # 管理者のみ
def post(self):
# 省略
'''エンドポイント：api/users/{id}'''
class UserAPI(Resource):
'''ユーザ取得'''
@jwt_required # 通常のトークン認証（ログイン中か）
def get(self, id):
# 省略
'''ユーザ編集'''
@self_or_admin_required_user # 本人または管理者
def patch(self, id):
# 省略
'''ユーザ削除'''
@admin_required # 管理者のみ
def delete(self, id):
# 省略
後回しにしていましたが、肝心のログイン処理はこのようになります。
from flask import request, jsonify
from flask_restful import Resource
from ..error import Abort
from ..database import db
from ..token import (
get_jwt_identity, create_access_token,
jwt_refresh_token_required, create_refresh_token,
set_access_cookies, set_refresh_cookies, unset_jwt_cookies
)
from ..models import User, UserSchema
from ..validators.auth import auth_login_schema
'''エンドポイント：api/auth/login'''
class AuthLoginAPI(Resource):
'''ログイン'''
def post(self):
input_data = request.json
errors = auth_login_schema.validate(input_data)
if errors:
Abort.error_4000(str(errors))
user = db.session.query(User).filter_by(email=input_data['email']).first()
if not user or not user.check_password(input_data['password']):
Abort.error_4010('IDもしくはパスワードに誤りがあります')
# Flask response object にする（dict object はダメ）
res = jsonify(UserSchema().dump(user))
# JWTの発行とクッキーへのセット(CSRFトークンも自動的にセットされる)
access_token = create_access_token(identity=user.id) # ユーザIDをセットする
set_access_cookies(res, access_token)
# 「ログインを保持」する場合、リフレッシュトークンもセットする
if input_data['remember'] == 1:
refresh_token = create_refresh_token(identity=user.id)
set_refresh_cookies(res, refresh_token)
return res
'''エンドポイント：api/auth/refresh'''
class AuthRefreshAPI(Resource):
'''トークン再発行'''
@jwt_refresh_token_required
def post(self):
current_user_id = get_jwt_identity() # ログイン時にセットしたユーザID
access_token = create_access_token(identity=current_user_id)
res = jsonify({'is_success': True})
set_access_cookies(res, access_token)
return res
'''エンドポイント：api/auth/logout'''
class AuthLogoutAPI(Resource):
'''ログアウト'''
def post(self):
res = jsonify({'is_success': True})
unset_jwt_cookies(res)
return res
以上のようにすると、SPA開発におけるログイン認証をRESTfulに実装できます。
config.pyにおけるFlask-JWT-Extendedの設定でいろいろ変わってきますので、実装するときはこちらをご確認ください。
DBのリレーション
最後に、SQLAlchemy・marshmallowを使って、テーブル同士のリレーションとリレーションに基づくレスポンスボディの整形についてご紹介したいと思います。
One to One
ArticleモデルとUserモデルのリレーションについて考えます。
１記事に対して１ユーザ（著者）が対応している場合、以下のようにします。
from datetime import datetime
from ..database import db, ma, fields
from . import UserSchema
class Article(db.Model):
__tablename__ = 'articles'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) # 外部キー
title = db.Column(db.String(191), nullable=True)
content = db.Column(db.Text, nullable=True)
is_public = db.Column(db.Tinyint(1), nullable=False, default=0)
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)
# One to One（1記事に対して、1つ）【スキーマ用なので、外部からプロパティをいじらない】
author = db.relationship('User', uselist=False, cascade='')
class ArticleSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Article
# UserSchemaに応じたネストされたデータを出力できる
author = fields.Nested(UserSchema)
'''
ArticleSchema.dump(article)の結果
{
"id": 1,
"author": {
"id": 1,
"name": "太郎",
"email": "user@example.com",
"is_admin": 0,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
},
"title": "太郎の話",
"content": "おれは生きる",
"is_public": 1,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
}
'''
db.ForeignKey('users.id')でUserモデルとのリレーションを定義します（DBにリレーションを知らせる）。
author = db.relationship('User', uselist=False, cascade='')とプロパティを定義することで、リレーションに基づいたUserモデルのデータを引っ張ってこれるようになります。つまり、リレーションに基づいたArticleモデルのデータがスキーマで必要のないUserモデル側には、プロパティを定義する必要がありません。
そして、このリレーションに基づいたSQLの実行はauthorにアクセスするときに行われる（デフォルト設定）ので、ViewファイルでArticleSchema.dump(articles, many=True)をしたときに１レコードずつリレーションに基づいたSQLが実行されます。つまり、SQL実行回数が増えるので処理が重くなる可能性が高いです。
ただ、ViewファイルでいろいろなSELECT処理実行とデータの結合などを行う必要がなくなり、他のエンドポイントの処理との対称性を保てるので、こちらを採用しました。
リレーションについて、詳しくはこちらの記事が参考になります。
→ SQLAlchemyのリレーションにおけるメソッドやパラメータについてのメモ
もう一度言いますが、リレーションに基づいたSQLの実行は該当のプロパティにアクセスするときに行われるので、記事取得のSELECT文とは完全に別であり、SELECT文を実行するときはリレーションに基づいたデータ（authorなど）についての情報はありません。つまり、Where句などに必要のないモデルはわざわざJOINする必要がありません。
補足ですが、uselist=Falseは1:1であることを明示していて、cascade=''は外部からUserモデルのデータを一切変更できないようにしています。
One to Many
ArticleモデルとTagモデルのリレーションについて考えます。
Tagモデルは今回記載しませんが、article_id = db.Column(db.Integer, db.ForeignKey('articles.id'), nullable=False)でArticleモデルとのリレーションを定義しています。
１記事に対して複数のタグが設定できる仕様です。つまり、同じarticle_idのレコードがTagテーブルに複数存在する状態です。
先ほどのapi/models/article.pyに処理を追加します。
# 省略
from . import UserSchema, TagSchema
class Article(db.Model):
# 省略
# One to Many（1記事に対して、複数）【記事からのリレーションが解かれると、自動的にDBから削除される】
tags = db.relationship('Tag', uselist=True, single_parent=True, cascade='all, delete-orphan')
def __init__(self, user_id, is_public):
self.user_id = user_id
self.is_public = is_public
class ArticleSchema(ma.SQLAlchemyAutoSchema):
# 省略
tags = fields.List(fields.Nested(TagSchema))
tag_count = fields.Function(lambda obj: len(obj.tags))
'''
ArticleSchema.dump(article)の結果
{
"id": 1,
"author": {
"id": 1,
"name": "太郎",
"email": "user@example.com",
"is_admin": 0,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
},
"title": "太郎の話",
"content": "おれは生きる",
"tag_count": 2,
"tags": [
{
"id": 1,
"name": "REST API"
},
{
"id": 2,
"name": "人生"
}
],
"is_public": 1,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
}
'''
uselist=Trueは1:Nであることを明示していて、single_parent=True, cascade='all, delete-orphan'で外部からTagモデルのデータを変更できるかつ、記事とのリレーションが解かれるとDBからタグが削除されるようになります。
また、cascade='delete'を指定すると、記事が削除されるタイミングでリレーション先も削除されるようになります。これを使うと、記事削除時の外部キーエラーを回避できます。（記事に対するコメントやいいねなど）
tag_count = fields.Function(lambda obj: len(obj.tags))では、出力フィールドを自作しています。objにはArticleモデルのインスタンスが代入されます。つまり、該当記事レコードに対してリレーションしているタグレコード（tagsプロパティ）の数を出力するようにしています。
他にもメソッドでフィールドを定義できたりします。詳しくは、公式をご確認ください。
Viewファイルでは、以下のようにして扱います。
from flask import request
from ..database import db
from ..token import get_jwt_identity
from ..models import Article, ArticleSchema, Tag
'''POSTリクエスト（実際はResourceクラスの定義、バリデーション実行など必要）'''
input_data = request.json
# 新しい記事のインスタンスを作成
article = Article(user_id=get_jwt_identity(), is_public=input_data['is_public'])
if 'title' in input_data:
article.title = input_data['title']
if 'content' in input_data:
article.content = input_data['content']
if 'tags' in input_data:
for tag_name in input_data['tags']:
tag = Tag(tag_name) # Tagインスタンスを作成（article_idはリレーションで自動補完される）
article.tags.append(tag) # Articleインスタンスのtagsリレーションに追加
# データベースに追加（タグ情報も一緒に）
db.session.add(article)
db.session.commit()
res = ArticleSchema().dump(article)
'''リレーションしているタグの削除'''
article.tags = [] # 編集系のリクエスト（PUTなど）のときに有用
db.session.commit()
他にも、現在のユーザがこの記事にいいねをしているかなどをスキーマで表現したいとき、contextを使用して外部からデータを渡して実装することができます。
'''modelファイル'''
class Article(db.Model):
# 省略
favorites = db.relationship('Favorite', uselist=True, cascade='delete')
class ArticleSchema(ma.SQLAlchemyAutoSchema):
# 省略
favorite_count = fields.Function(lambda obj: len(obj.favorites))
favorited_by_user = fields.Function(lambda obj, context: context['current_user_id'] in [favorite_obj.user_id for favorite_obj in obj.favorites])
'''Viewファイル'''
article = db.session.query(Article).filter_by(id=id).first()
# contextを利用して、現在のユーザ情報をスキーマに渡す
schema = ArticleSchema()
schema.context = {'current_user_id': get_jwt_identity()}
res = schema.dump(article)
'''
{
"id": 1,
"author": {
"id": 1,
"name": "太郎",
"email": "user@example.com",
"is_admin": 0,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
},
"title": "太郎の話",
"content": "おれは生きる",
"tag_count": 2,
"tags": [
{
"id": 1,
"name": "REST API"
},
{
"id": 2,
"name": "人生"
}
],
"favorite_count": 6,
"favorited_by_user": true,
"is_public": 1,
"created_at": "2019-08-24T14:15:22Z",
"updated_at": "2019-08-24T14:15:22Z"
}
'''
まとめ
最初は、セクションごとに記事を書こうと思ったのですが、面倒くさくなってしまいました。
結果、膨大な量になってしまいましたが、これでも全部は紹介しきれていないです。
簡単のため、「省略」と書いていない箇所も省略していたりするので、動かなかったりしたら申し訳ございません。あくまで、ご紹介した内容すべてを適用すると「男前に」なったよって話です。
ただ、FlaskはもちろんPythonも初心者なので、変な書き方をしているかもしれません。なにかございましたら、ご指摘のほどよろしくお願いします。