API仕様書を作らなくてもいい！？簡単にAPIを作成する方法

この記事はニフティグループ Advent Calendar 2018の24日目の記事です。
昨日は@taka_masaさんの今年イチの脆弱性でした。
なかなか恐ろしい脆弱性・・・

#はじめに

いきなりですが、API仕様書を作成したりするのって結構面倒で手間がかかりますよね。
Excel等で作成すると実際にドキュメント通りの結果が返ってくるのかが担保できなかったり、
プログラムの仕様を変更したときに仕様書も合わせて修正したりしないといけなかったり・・・
最近では、Swagger等のAPIドキュメンテーションツールが整備されているので以前よりは作成コストは減りましたが、
可能であればコーディングをすればそれに合わせた仕様書を自動で作成してほしい・・・
それを実現できるのが今から紹介するFlask-RESTPlusです。

#Flask-RESTPlus
Flask-RESTPlusは、REST APIを素早く構築するためのFlaskの拡張ツールです。
特徴としては、コード内にデコレータでAPIに関する記述をすると自動でSwaggerUIが生成されるのでAPI仕様書を作成する必要がありません。
また、コードにAPIドキュメントを書くのでコードとドキュメントの内容に違いが発生しづらくなります。

コードを書く→SwaggerUI(API仕様書)が作られる→SwaggerUI上でテストする

Flask-RESTPlusを利用するとこのサイクルを高速で回せるので開発スピードも向上します。

#Flask-SQLAlchemy
Flask-SQLAlchemyはpythonのORMであるSQLAlchemyをFlask向けにカスタマイズされたFlaskの拡張機能です。DBのセッションを特に意識することなく簡素なコードでDBとのやり取りが可能であるのが特徴です。

#環境
Python 3.6
MariaDB 5.5
IDE Pycharm 2018.3.1

#Step1. 必要なパッケージのインストールと簡単なサンプルを動かしてみる
パッケージをインストール
pip install flask,flask-restplus,flask-sqlalchemy, pymysql

公式のQuickStartに従って簡単なプログラムを作成します。

from flask import Flask
from flask_restplus import Resource, Api

app = Flask(__name__)
api = Api(app)

@api.route('/hello')
class HelloWorld(Resource):
    def get(self):
        return {'hello': 'world'}

if __name__ == '__main__':
    app.run(debug=True)

実行してみると簡易APIサーバが立ち上がるのでlocalhost:5000にブラウザでアクセスしてみます。
アクセスしてみるとSwaggerUIが確認できるので、右上にTry it outを押して実際に試してみます。

するとExecuteボタンが出てくるので押すと実行されてレスポンスが表示されます。

#Step2　APIモデルとDBモデルを定義する
無事にサンプルが動かせたので簡単なAPIを作っていきます。
今回は、社員情報を取得するAPIを試しに作っていおうとおもいます。

##DBモデルの定義
まずは、DBのモデルを定義していきます。
先ほどのコードに追記をして以下のようにDBモデルを作成します。

from flask import Flask
from flask_restplus import Resource, Api, Namespace, fields
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class Worker(db.Model):
    __tablename__ = "t_worker"

    id = db.Column(db.String(length=32), primary_key=True)
    family_name = db.Column(db.String(length=3), nullable=False)
    first_name = db.Column(db.String(length=32), nullable=False)
    age = db.Column(db.Integer, nullable=True)
    company_id = db.Column(db.ForeignKey('t_company.id'), nullable=False)


class Company(db.Model):
    __tablename__ = "t_company"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    company_name = db.Column(db.String(length=128), nullable=False)
    city_id = db.Column(db.Integer, db.ForeignKey('t_city.id'), nullable=True)
    worker = db.relationship('Worker', backref="t_company")


class City(db.Model):
    __tablename__ = "t_city"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    city_name = db.Column(db.String(length=128), nullable=False)
    company = db.relationship("Company", backref="t_company")


app = Flask(__name__)
api = Api(app, title="サンプル社員情報取得API", description="これはサンプルです")


@api.route('/hello')
class HelloWorld(Resource):
    def get(self):
        return {'hello': 'world'}


if __name__ == '__main__':
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://{USER}:{PASS}@{HOST}/{DB_NAME}?charset=utf8'
    db.init_app(app)
    db.create_all(app=app)
    app.run(debug=True)

上のアプリケーションを実行してみると定義したDBが自動で作成されます。
Pycharmのダイアグラム表示機能を使用して確認してみると以下のようになっています。

##APIモデルを定義する
次にAPIのモデルを定義していきます。
モデルを定義すると同時にAPIの名前空間を定義してあげます。
また、定義した名前空間を使ってルーティングして、軽い処理を書いていきます。
先ほどのコードに下記のコードを追記して、HelloWorldを置き換えます。

worker_namespace = Namespace('worker', description='社員関連のエンドポイント')

# APIモデル
city = worker_namespace.model('City', {
    'city_name': fields.String(
        description='都市名'
    )
})

company = worker_namespace.model('Company', {
    'company_name': fields.String(
        description="会社名"
    ),

    'city': fields.Nested(city, readonly=True)
})


worker = worker_namespace.model('Worker', {

    'family_name': fields.String(
        required=True,
        example="田中",
        description="名字"
    ),

    'first_name': fields.String(
        required=True,
        example="太郎",
        description="下の名前"
    ),

    'age': fields.Integer(
        example=25,
        description="年齢"
    ),

    'company': fields.Nested(company, readonly=True)

})

# 略

@worker_namespace.route('/')
class WorkerRoute(Resource):
    def get(self):
        return

# 略

if __name__ == '__main__':
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://{USER}:{PASS}@{HOST}/{DB_NAME}?charset=utf8'
    db.init_app(app)
    db.create_all(app=app)
    api.init_app(app)
    app.run(debug=True)

このあたりで定義している、requiredやreadonlyでSwaggerUIでの振る舞いを変えることができます。
required=Trueで入力必須項目にすることができ、readonly=Trueを定義してあげると読み込み専用のオブジェクトになります。それをすると、SwaggerUI上でPOST時のbodyフィールドに表示されなくなります。exampleを付与するとPOST時のbodyフィールドの初期値として格納することができます。
また、APIモデル内でpattern=を使って正規表現を利用したバリデーションをすることも可能です。

ここまで準備ができたら、いよいよAPIの処理とドキュメントを書いていきます。

#Step3. デコレータをつけてAPIのドキュメントを作成する
Flask-RESTPlusの公式ドキュメントを参考にデコレータを付与してドキュメントを作成していきます。

@worker_namespace.route('/')
class WorkerRoute(Resource):
    @worker_namespace.response(200, "Worker_Information", model=worker)
    def get(self):
        return

    @worker_namespace.expect(worker)
    @worker_namespace.response(201, "Worker Created")
    def post(self):
        return

    @worker_namespace.response(204, "Worker Deleted")
    def delete(self):
        return

この状態でアプリケーションを実行してブラウザを立ち上げると、先ほど作成したモデルをベースとしたAPIドキュメントが作成されています。

このように、対象の処理にデコレータを付与してあげるだけで簡単にAPIドキュメントが作れます。

#Step4. APIの処理を書く
あとは、APIの処理を書いてあげるだけです。
Flask-SQLAlchemyを使ったクエリの書き方は公式ドキュメントを参考にしてください。

前準備で先ほど作ったDBにデータを入れておきます。

t_city

t_company

そしてコードは最終的に下記のように書きました。

test_app.py

from flask import Flask, request
from flask_restplus import Resource, Api, Namespace, fields, abort
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

app = Flask(__name__)
api = Api(title="サンプル社員情報取得API", description="これはサンプルです")
worker_namespace = Namespace('worker', description='社員関連のエンドポイント')
api.add_namespace(worker_namespace)


class Response(object):
    def __init__(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)


class Worker(db.Model):
    __tablename__ = "t_worker"

    id = db.Column(db.String(length=32), primary_key=True)
    family_name = db.Column(db.String(length=32), nullable=False)
    first_name = db.Column(db.String(length=32), nullable=False)
    age = db.Column(db.Integer, nullable=True)
    company_id = db.Column(db.ForeignKey('t_company.id'), nullable=False)


class Company(db.Model):
    __tablename__ = "t_company"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    company_name = db.Column(db.String(length=128), nullable=False)
    city_id = db.Column(db.Integer, db.ForeignKey('t_city.id'), nullable=True)
    worker = db.relationship('Worker', backref="t_company")


class City(db.Model):
    __tablename__ = "t_city"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    city_name = db.Column(db.String(length=128), nullable=False)
    company = db.relationship("Company", backref="t_company")


# APIモデル
city = worker_namespace.model('City', {
    'city_name': fields.String(
        description='都市名'
    )
})

company = worker_namespace.model('Company', {
    'company_name': fields.String(
        description="会社名"
    ),

    'city': fields.Nested(city, readonly=True)
})

worker = worker_namespace.model('Worker', {

    'family_name': fields.String(
        required=True,
        example="田中",
        description="名字"
    ),

    'first_name': fields.String(
        required=True,
        example="太郎",
        description="下の名前"
    ),

    'age': fields.Integer(
        example=25,
        description="年齢"
    ),

    'company_id': fields.Integer(
        example=1,
        description="会社コード"
    ),

    'company': fields.Nested(company, readonly=True)

})


@worker_namespace.route('/<string:worker_id>')
class WorkerRoute(Resource):
    @worker_namespace.marshal_with(worker)
    def get(self, worker_id):
        try:
            worker_info = db.session.query(Worker, Company, City).outerjoin(Company,
                                                                            Worker.company_id == Company.id).outerjoin(
                City, Company.city_id == City.id).filter(Worker.id == worker_id).one_or_none()
            if worker_info is not None:
                city_data = Response(city=worker_info.City)
                worker_info.Company.__dict__.update(city_data.__dict__)
                company_data = Response(company=worker_info.Company)
                worker_info.Worker.__dict__.update(company_data.__dict__)
                return worker_info.Worker
            else:
                return
        except Exception:
            return abort(500, "Internal Server Error")

    @worker_namespace.expect(worker)
    @worker_namespace.response(201, "Worker Created")
    def post(self, worker_id):
        try:
            payload = request.json
            family_name = payload["family_name"]
            first_name = payload["first_name"]
            age = payload["age"] if payload["age"] is not None else None
            company_id = payload["company_id"]
            post_obj = Worker(id=worker_id, family_name=family_name, first_name=first_name, age=age,
                              company_id=company_id)
            db.session.add(post_obj)
            db.session.commit()
            return {"message": "Worker Created"}, 201
        except Exception:
            return abort(500, "Internal Server Error")

    @worker_namespace.response(204, "Worker Deleted")
    def delete(self, worker_id):
        try:
            db.session.query(Worker).filter(Worker.id == worker_id).delete()
            db.session.commit()
            return 204
        except Exception:
            return abort(500, "Internal Server Error")


if __name__ == '__main__':
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://{USER}:{PASS}@{HOST}/{DB_NAME}?charset=utf8'
    api.init_app(app)
    db.init_app(app)
    db.create_all(app=app)
    app.run(debug=True)

アプリケーションを起動してみてブラウザでSwaggerUIを確認してみます。

POSTしてユーザを登録してみます。

201のレスポンスが返ってきたのでユーザが作られたようです。

作成したユーザの情報をGETしてみましょう。

登録したユーザの情報が取れました。

DELETEしてみます。
無事にデータが削除されたようです。

もう一度GETしてみます。
ユーザ情報がないので中身が空で返ってきていますね。

#最後に
これだけのコード量でAPIを簡単に作成する上にSwaggerUIまで自動で作ってくれます。
簡単に作れるだけでなく、保守性も高くなりますね。
また、スピードが求められる今の時代において、少ないコード量で実現できるというのはとても大事ですよね。
素早く安全に、APIを作るためにも今後もFlask-RESTPlusを活用していきたいと思います。

いよいよ明日は最終日。@jimmysharpさんの記事です。お楽しみに。