flask+marshmallowなAPIをOpenAPIを使った仕組みに移行した話

こんにちは！株式会社fluct DATA STRAP事業本部エンジニアの宮前です。自分はDATA STRAPというレポート運用サポートシステムの開発運用保守などを対応しているエンジニアです。

今回はそのDATA STRAPの既存のAPIの仕組みをOpenAPIを使った新しいAPIの仕組みに順次移行している話をしたいと思います。

きっかけはAPIとmockの乖離

なぜOpenAPIを利用したAPIに移行しているかというと、既存のAPIはAPIとフロント開発用のmockのschemaを別で定義する必要があることが主な理由です。

既存APIはmarshmallowというライブラリを利用していて、フロントからPOSTされるjsonとAPI内で使うclassの相互変換 & validationなどに使っています。

以下はmarshmallowで作成したschemaです。

class UserSchema(Schema):
    id = fields.Int(required=True, dump_only=True)
    name = fields.Str(required=True, validate=validate_name)

    @post_load
    def make_object(self, data, **kwargs):
        return UserStruct(**data)

    @staticmethod
    def create(req_json: dict) -> UserStruct:
        try:
            return UserSchema().load(req_json)
        except ValidationError as e:
            raise MarshmallowExtException(e.messages)

一方でmockの作成には、FastAPIを利用しています。

mockの実装は以下のようになっています

class UsersAPI:
    @app.post(
        tags=["users"],
        path="/v1/users",
        status_code=201,
        response_model=User,
        description="ユーザを登録",
    )
    def user_create(p: PostUser):
        return {"user": User(id=4, name=p.name)}

class User(BaseModel):
    id: int
    name: str

class PostUser(BaseModel):
    name: str

上記のようにAPIとmockで同じUserというschemaを別々に作る必要があります。これによって、APIとmockの実装の乖離が起き、mockを基に作成したフロントエンドと動いているAPIの連携がうまくいかないということが多かったのが既存APIの問題でした。

他にもmock用のschemaを別で定義するのが大変だったり、schemaのvalidationがmarshmallowで書かれているためpythonに大きく依存している点も改善したい点として挙げられました。

そこで、もっと良い仕組みにできないかという話し合いがチーム内で行われた結果、schema定義からAPI, mockのどちらも生成できるOpenAPIを利用して作成する方針になりました。

OpenAPIを利用したAPI

作成する際に意識した点としては、できるだけ早く作って利用する、作り込みすぎないことです。新APIは動かしていかないと辛いポイントが見えてこないと思ったのでやりすぎないように進めました。

ここからは、作成したOpenAPIを利用したAPIの大まかな処理の流れを紹介します。

まずはOpenAPI Specification(openapi.yaml)にschema定義を書きます。

paths:
  /v1/users:
    post:
      tags:
        - users
      operationId: rpz.api.views.users.create
      summary: ユーザの作成
      description: ユーザを作成します
      requestBody:
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PostUser'
        required: true
      responses:
        '201':
          description: 作成成功
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'

components:
  schemas:
    PostUser:
      description: ユーザの作成用スキーマ
      type: object
      properties:
        name:
          description: ユーザの名前
          type: string
      required:
        - name

    User:
      type: object
      properties:
        id:
          type: integer
          readOnly: true
        name:
          type: string
      required:
        - name

※ operationIdについては後述します

そしてこのyamlからOpenAPI Generatorを利用してmodelを生成します。

生成されたmodelは以下のような感じです。

# coding: utf-8

from __future__ import absolute_import
from datetime import date, datetime  # noqa: F401

from typing import List, Dict  # noqa: F401

from openapi_server.models.base_model_ import Model
from openapi_server import util


class User(Model):
    """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).

    Do not edit the class manually.
    """

    def __init__(self, id=None, name=None):  # noqa: E501
        """User - a model defined in OpenAPI

        :param id: The id of this User.  # noqa: E501
        :type id: int
        :param name: The name of this User.  # noqa: E501
        :type name: str
        """
        self.openapi_types = {
            'id': int,
            'name': str
        }

        self.attribute_map = {
            'id': 'id',
            'name': 'name'
        }

        self._id = id
        self._name = name

    @classmethod
    def from_dict(cls, dikt) -> 'User':
        """Returns the dict as a model

        :param dikt: A dict.
        :type: dict
        :return: The User of this User.  # noqa: E501
        :rtype: User
        """
        return util.deserialize_model(dikt, cls)

    @property
    def id(self):
        """Gets the id of this User.


        :return: The id of this  User.
        :rtype: int
        """
        return self._id

    @id.setter
    def id(self, id):
        """Sets the id of this  User.


        :param id: The id of this  User.
        :type id: int
        """

        self._id = id

    @property
    def name(self):
        """Gets the name of this  User.


        :return: The name of this  User.
        :rtype: str
        """
        return self._name

    @name.setter
    def name(self, name):
        """Sets the name of this  User.


        :param name: The name of this  User.
        :type name: str
        """
        if name is None:
            raise ValueError("Invalid value for `name`, must not be `None`")  # noqa: E501

        self._name = name

そして上記の、frontendとやり取りするためのmodelと既存のORM(SQLAlchemy) classを変換する関数を作成し、controllerで利用するようにしました。

変換関数は以下のような実装です。

def convert2user(user: rpz.models.User) -> openapi_server.models.User:
    return openapi_server.models.User(
        id=user.id,
        name=user.name,
    )

controllerの実装も後述します。

mockはprismを利用し、openapi.yamlから立ち上がるようにしました。

openapi.yamlによってAPIが利用するmodelとmockが自動で生成されるので、API, mock間の乖離が無くなり、不要な手戻りも発生しなくなりました。めでたい！

移行する際の工夫

既存のAPIから新APIに一気に全てを移行するのは大変なのでどちらも共存できるようにしました。そのために工夫した点を一部紹介します。

controller, routingの共存

既存のAPI基盤はflaskを利用しておりルーティングなどもflaskの機能を使っています。

routingは以下のように行っています。

def json_api_routing(app: Flask):
    app.add_url_rule(
        "/v1/users",
        view_func=UserCreate.as_view("company_user_create"),
    )

view_funcで指定したclassから対応するHTTPメソッドと同名の関数が呼び出されるようになっています。

controllerは以下のような感じです。

user_schema = UserSchema()

class UserCreate(APIMethodView):
    def post(self):
        x = user_schema.create(request.json)
        user = UserRepository.create(x.name)
        ret = self.build_response(user)
        return make_response(jsonify(ret), 201)

    def build_response(self, user: User) -> dict:
        return {"user": user_schema.dump(user)}

継承しているAPIMethodViewではflaskのMethodViewを継承し、認証などの処理を追加しています。

冒頭でも触れましたが、UserSchemaはmarshmallowで作成しており、jsonとclassの相互変換やvalidationを行っています。

class UserSchema(Schema):
    id = fields.Int(required=True, dump_only=True)
    name = fields.Str(required=True, validate=validate_name)

    @post_load
    def make_object(self, data, **kwargs):
        return UserStruct(**data)

    @staticmethod
    def create(req_json: dict) -> UserStruct:
        try:
            return UserSchema().load(req_json)
        except ValidationError as e:
            raise MarshmallowExtException(e.messages)

次に、作成した新APIのroutingとcontrollerについて紹介します。

新APIのopenapi.yamlからOpenAPI Generatorを用いて様々なコードを自動生成できるのですが、実際に自動生成するコードはmodelと細かいutilファイルのみにしました。既存の仕組みに乗れるところは乗れるようにしたいのでその他のcontrollerなどは生成対象外としています。

新APIのroutingと入出力値のvalidationにはconnexionを利用しています。validationライブラリはconnexionの他にもbravado-coreというライブラリも導入検討していましたが、OpenAPI3.0非対応だったため導入を見送りました。まとめの章で少し触れていますが、yamlを自動生成するようにすれば導入するかもしれません。

connexionでroutingするにはopenapi.yamlで指定するoperationIdで処理する関数を指定する必要があります。

以下が新APIのcontrollerです。

def create():
    return UserCreate().dispatch_request()

class UserCreate(APIMethodView):
    def post(self):
        post_user = PostUser.from_dict(
            connexion.request.get_json()
        )
        user = UserRepository.create(post_user)
        ret = self.build_response(user)
        return make_response(jsonify(ret), 201)

    def build_response(self, user: User) -> dict:
        return convert2user(user).to_dict()

従来の基盤のroutingでは、APIMethodViewが継承しているMethodViewのdispatch_request()が暗黙的に呼ばれていましたが、connexionを利用するにあたって処理を担当する関数を指定する必要があるので直接dispatch_request()を呼び出す関数を新しく作成しています。

以下の設定でflaskのroutingとconnexionのroutingどちらも利用するようにしました。既存のroutingはそのまま使い、新APIはconnexionでroutingするようにしています。

connexion_app = connexion.FlaskApp(__name__, specification_dir="./")
app = connexion_app.app

def init_app() -> Flask:
    ~~(省略)~~
    json_api_routing(app)  # 既存のrouting, flaskのapp.add_urlなどを利用している
    connexion_init(connexion_app)  # connexionのrouting
    app.app_context().push()
    return app

def connexion_init(con_app):
    con_app.add_api(
        "openapi.yaml",
        validate_responses=True,
    )
    con_app.add_error_handler(ProblemException, render_exception)

これで従来の基盤の構成に新しいroutingとschemaをそのまま乗せることができました。

mockの共存

prismのvalidation proxyという機能を使うことにより実現しました。

validation proxyは、proxy先が501 not implementedを返すとprismがmockしたレスポンスを返すというものです。

今回は既存のmockが知らないroutingは501を返すようにし、代わりにprismがレスポンスを返すようにしました。

@app.exception_handler(404)
def handle_404(_req, _exc):
    return Response(status_code=501)

↑のように旧mockが知らないURLの場合、501を返すようにしています。

prism側は、prism proxy -p 8000 app_py/rpz/openapi.yaml http://localhost:8001 といった感じでportを分けて立ち上げています。

まとめ

以上のような取り組みにより、

APIとmock間の乖離がなくなり、不要な手戻りが発生しなくなった
modelの自動生成により、コードを書く量が減った
旧APIと共存しているので既にある機能についてはAPI, mock共に急いで移行する必要がない
- 現在は既存APIに手を入れる必要が生じた時にその部分のAPIを新APIに移行するように進めている

などが実現できました。

残った課題として、yamlを人の手で作成、レビューするのは大変なのでStoplightなどのIDEを利用するか、Protocol Buffers等からyamlを生成するか*1などの方法をいずれ取ろうと考えています。