概要
Python FlaskでREST APIの開発に挑戦してみる。
blueprintでURLマッピングを行い、Flask-RESTfulでAPIを実装、
Cerberusでバリデーションをし、SQLAlchemyでDBとやり取りすることにする。
準備
Python + uWSGI + nginx で事前にWebサーバーの立ち上げをしておいた。 また、
pip3 install blueprint Cerberus Flask-RESTful Flask-SQLAlchemy Flask-SQLAlchemy-Session mysqlclient SQLAlchemy marshmallow marshmallow-sqlalchemy
で追加のライブラリをインストールした。
ファイル構成
app.py
uwsgi.ini
controller/
+ __init__.py
+ auth.py # APIその1 (URL: /api/auth/～)
+ hoge.py # APIその2 (URL: /api/hoge/～)
engine/
+ __init__.py
+ read.py # 読み取り用DB接続
+ write.py # 書き込み用DB接続
model/
+ __init__.py
+ user.py # userテーブルとやり取りするモデル
validator/
+ __init__.py
+ extended.py # 追加バリデーション
app.pyの作成
リクエスト時のエントリーポイント。このファイルはblueprintを活用して振り分けに専念し、肥大化しないようにする。
from flask import Flask
from controller import auth, hoge
app = Flask(__name__)
app.register_blueprint(auth.app, url_prefix = '/api')
app.register_blueprint(hoge.app, url_prefix = '/api')
if __name__ == '__main__':
app.run(debug = True)
DB処理の作成
engine/write.py (read.pyも同様に)
一応書き込みと読み込みでDBが分かれているケースを想定。
from flask import current_app
from flask_sqlalchemy_session import flask_scoped_session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class WriteEngine(object):
def __init__(self):
username = 'hogehoge'
password = 'mogemoge'
hostname = 'localhost'
dbname = 'rest'
charset = 'utf8mb4'
url = 'mysql+mysqldb://{}:{}@{}/{}?charset={}'.format(username, password, hostname, dbname, charset)
self.engine = create_engine(url, echo = True)
class WriteSession(WriteEngine):
def __init__(self):
super().__init__()
self.session = flask_scoped_session(sessionmaker(bind = self.engine), current_app)
model/user.py
ここでは user テーブルがあるとして、そのテーブルのモデルクラスを作成する。
from sqlalchemy import text
from sqlalchemy.dialects.mysql import INTEGER as Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.functions import current_timestamp
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
db = SQLAlchemy()
ma = Marshmallow()
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = db.Column(Integer(unsigned = True), primary_key = True)
email = db.Column(db.String(128), nullable = False, unique = True)
name = db.Column(db.String(64), nullable = False)
password = db.Column(db.String(64), nullable = False)
created_at = db.Column(db.DateTime, nullable = False, server_default=current_timestamp())
updated_at = db.Column(db.DateTime, nullable = False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'))
class UserSchema(ma.Schema):
class Meta:
fields = ('id', 'email', 'name', 'created_at', 'updated_at')
バリデーション追加処理の作成
Cerberusで最初から付いてくるバリデーションでは物足りないので追加のバリデーションを入れる。
メッセージの日本語化は中途半端だが必要に応じて追々埋めていくことにする。
import re
from cerberus import Validator
from cerberus.errors import BasicErrorHandler
class ValidatorExtended(Validator):
def _validate_valid_email(self, email, field, value):
"""
The rule's arguments are validated against this schema: {'type': 'boolean'}
"""
if (email and not re.match('^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', value)):
self._error(field, '正しいメールアドレスではありません。')
class JapaneseErrorHandler(BasicErrorHandler):
def __init__(self, tree = None):
super(JapaneseErrorHandler, self).__init__(tree)
self.messages = {
0x00: "{0}",
0x01: "document is missing",
0x02: "必須項目です。",
0x03: "不明な項目が指定されています。",
0x04: "'{0}'は必須項目です。",
0x05: "depends on these values: {constraint}",
0x06: "{0} must not be present with '{field}'",
0x21: "'{0}' is not a document, must be a dict",
0x22: "必須項目です。",
0x23: "必須項目です。",
0x24: "must be of {constraint} type",
0x25: "must be of dict type",
0x26: "length of list should be {constraint}, it is {0}",
0x27: "{constraint}文字以上入力してください。",
0x28: "{constraint}文字以内で入力してください。",
0x41: "value does not match regex '{constraint}'",
0x42: "{constraint}以上の値を入力してください。",
0x43: "{constraint}以下の値を入力してください。",
0x44: "{value}は指定できません。",
0x45: "{0}は指定できません。",
0x46: "{value}は指定できません。",
0x47: "{0}は指定できません。",
0x48: "missing members {0}",
0x61: "field '{field}' cannot be coerced: {0}",
0x62: "field '{field}' cannot be renamed: {0}",
0x63: "field is read-only",
0x64: "default value for '{field}' cannot be set: {0}",
0x81: "mapping doesn't validate subschema: {0}",
0x82: "one or more sequence-items don't validate: {0}",
0x83: "one or more keys of a mapping don't validate: {0}",
0x84: "one or more values in a mapping don't validate: {0}",
0x85: "one or more sequence-items don't validate: {0}",
0x91: "one or more definitions validate",
0x92: "none or more than one rule validate",
0x93: "no definitions validate",
0x94: "one or more definitions don't validate"
}
コントローラーの作成
controller/auth.py
from flask import Blueprint, request
from flask_restful import Resource, Api
from validator.extended import ValidatorExtended, JapaneseErrorHandler
from engine.write import WriteSession
from model.user import User, UserSchema
from util.crypt import UtilCrypt
from util.token import UtilToken
app = Blueprint('auth', __name__)
api = Api(app)
# /api/auth/login でアクセスされるAPIの処理
class AuthLoginResource(Resource):
# ユーザーのログインチェック処理 (POSTメソッド)
def post(self):
# バリデーション設定
schema = {
'email': {
'type': 'string',
'required': True,
'empty': False,
'maxlength': 128,
'valid_email': True
},
'password': {
'type': 'string',
'required': True,
'empty': False,
'maxlength': 32
},
}
validator = ValidatorExtended(schema, error_handler = JapaneseErrorHandler())
# バリデーション失敗
if not validator.validate({} if request.json is None else request.json):
return {'result': False, 'errors': validator.errors}
# バリデーション後のリクエストパラメーターを受け取る (バリデーションで加工処理を入れていた場合、加工分が反映される)
params = validator.document
# DBに重複レコードが無いか確認
user = WriteSession().session.query(User.id, User.name, User.password).filter(User.email == params['email']).first()
if (user is None or `パスワードが一致しない (ロジックは各々での実装)`):
return {'result': False, 'errors': {'email': 'ユーザーが存在しないかパスワードが一致しません。'}}
return {'result': True, 'token': `ログイントークン発行処理 (ロジックは各々での実装)`}
api.add_resource(AuthLoginResource, '/auth/login')
今回は、パラメーターはJSON形式で受け取る方式とした。
同様に他のコントローラーも作り込みをしていく。
メソッド名をget, put, deleteにすれば、他のHTTPメソッドの実装も可能な模様。
動作確認
curl -H 'Host: ホスト名' -H 'Content-Type:application/json' -d '{"email":"hoge@sample.jp","password":"hogehoge"}' -k https://localhost/api/auth/login
結果
{"result":true,"token":"xxxxyyyy"}
正しくやり取りできることが確認できた。