速くて軽量なAPI、Starletteを使ってGCPにデプロイする　Part2

今回すすめるトピック

さらっとソースの構造を確認していきましょう。実行するmain.py以外はsrcディレクトリに配置しています。Git,Docker,Cloudbuildのビルドで除外設定をしているのがxxx ignoreファイルです。
ほぼコンテナでインフラも運用しているので、どのプロジェクトもDocker,CloudBuildの設定が入っています。この記事の目的も最終はGCPでクラウドランで稼働させることなので、CloudBuildにCI/CDとデプロイのステップを書きます。
PipfileはPythonの仮想環境を作るライブラリで、DevとProdで依存関係を分けてインストールするのが楽なので使っています。最近はどうかわからないですけど、pipのバージョン最新にすると依存関係でうまくいかないことがあるので、19.23でバージョンは固定しています。あとはGCPでAppengineや、Cloud Functionsを使う場合はrequirements.txtが必要なので、pipenv lock -r > requirements.txtで一応作って置いておきます。あとはテスト用にpytest.ini, tests.pyといった構成になっています。

エンドポイントを追加する

Gitのソースが手元にない方はまずはクローンして、環境構築のPart1を進めてください。

前回の続きの方はアプリと依存環境を起動しましょう。とくにPostgreSQLが起動していないことはよくあるので、うまくいかないときはコマンドで確認してみてください。

# DockerコンテナでPostgreSQLが起動しているか確認
(starlette-cloudrun) __ksh__@macbook-pro starlette-cloudrun % docker ps
CONTAINER ID        IMAGE                  COMMAND                  CREATED             STATUS              PORTS                    NAMES
ef2168e96de4        postgres:11.6-alpine   "docker-entrypoint.s…"   2 days ago          Up 5 seconds        0.0.0.0:5432->5432/tcp   starlette-cloudrun_postgres_1

準備がよろしければ早速エンドポイントを追加して、モデルデータを返すAPIを作ってみましょう。

# main.py 50,59

@app.route('/v1/robots/repair', methods=['POST'])
async def repair(request: Request) -> str:
    form = await request.form()
    values = {'robot_id': int(form['robot_id'])}
    robot = await store.Robot.retrieve(values)
    logger.debug(robot)

    task = BackgroundTask(task_repair, robot)
    message = {'message': f'Start repairing {robot.name}.'}
    return JSONResponse(message, background=task)

# TODO: jsonでモデル情報をを返すAPIを作る
@app.route('/v1/robots/{robot_id:int}')
async def retrieve_robot(request: Request) -> str:
    robot_id = request.path_params["robot_id"]
    logger.debug(robot_id)

    values = {'robot_id': robot_id}
    robot = await store.Robot.retrieve(values)
    logger.debug(robot)

    # TODO: あとでバックグラウンドタスクを追加する時に使うのでコメントアウト
    # task = BackgroundTask(task_count_ref, robot)
    message = {'message': f'Robot is {robot.name}.'}
    return JSONResponse(message)

リクエストパスから値を取得したい場合はrequest.path_paramsから指定したpathキーで取り出します。Requestオブジェクトの継承元がHttp関連でRequestはasgiの実装になっています。注入する値はこれらの実装見ればだいたい分かりますので参照してください。

# starlette.requests.py

class HTTPConnection(Mapping):
    """
    A base class for incoming HTTP connections, that is used to provide
    any functionality that is common to both `Request` and `WebSocket`.
    """

    def __init__(self, scope: Scope, receive: Receive = None) -> None:
        assert scope["type"] in ("http", "websocket")
        self.scope = scope

    def __getitem__(self, key: str) -> str:
        return self.scope[key]

    def __iter__(self) -> typing.Iterator[str]:
        return iter(self.scope)

    def __len__(self) -> int:
        return len(self.scope)

    @property
    def app(self) -> typing.Any:
        return self.scope["app"]

    @property
    def url(self) -> URL:
        if not hasattr(self, "_url"):
            self._url = URL(scope=self.scope)
        return self._url

    @property
    def base_url(self) -> URL:
        if not hasattr(self, "_base_url"):
            base_url_scope = dict(self.scope)
            base_url_scope["path"] = "/"
            base_url_scope["query_string"] = b""
            base_url_scope["root_path"] = base_url_scope.get(
                "app_root_path", base_url_scope.get("root_path", "")
            )
            self._base_url = URL(scope=base_url_scope)
        return self._base_url

    @property
    def headers(self) -> Headers:
        if not hasattr(self, "_headers"):
            self._headers = Headers(scope=self.scope)
        return self._headers

    @property
    def query_params(self) -> QueryParams:
        if not hasattr(self, "_query_params"):
            self._query_params = QueryParams(self.scope["query_string"])
        return self._query_params

    @property
    def path_params(self) -> dict:
        return self.scope.get("path_params", {})

    @property
    def cookies(self) -> typing.Dict[str, str]:
        if not hasattr(self, "_cookies"):
            cookies = {}
            cookie_header = self.headers.get("cookie")
            if cookie_header:
                cookie = http.cookies.SimpleCookie()  # type: http.cookies.BaseCookie
                cookie.load(cookie_header)
                for key, morsel in cookie.items():
                    cookies[key] = morsel.value
            self._cookies = cookies
        return self._cookies

    @property
    def client(self) -> Address:
        host, port = self.scope.get("client") or (None, None)
        return Address(host=host, port=port)

    @property
    def session(self) -> dict:
        assert (
            "session" in self.scope
        ), "SessionMiddleware must be installed to access request.session"
        return self.scope["session"]

    @property
    def auth(self) -> typing.Any:
        assert (
            "auth" in self.scope
        ), "AuthenticationMiddleware must be installed to access request.auth"
        return self.scope["auth"]

    @property
    def user(self) -> typing.Any:
        assert (
            "user" in self.scope
        ), "AuthenticationMiddleware must be installed to access request.user"
        return self.scope["user"]

    @property
    def state(self) -> State:
        if not hasattr(self, "_state"):
            # Ensure 'state' has an empty dict if it's not already populated.
            self.scope.setdefault("state", {})
            # Create a state instance with a reference to the dict in which it should store info
            self._state = State(self.scope["state"])
        return self._state

    def url_for(self, name: str, **path_params: typing.Any) -> str:
        router = self.scope["router"]
        url_path = router.url_path_for(name, **path_params)
        return url_path.make_absolute_url(base_url=self.base_url)

それではAPIへリクエストを送ってみましょう。

(starlette-cloudrun) __ksh__@macbook-pro starlette-cloudrun % curl http://127.0.0.1:5000/v1/robots/1
{"message":"Robot is GUIDO."}%     

DEBUG LOG...

DEBUG:    1
DEBUG:    Robot(id=1, name='GUIDO', created_at=datetime.datetime(2020, 1, 1, 0, 0))
INFO:     127.0.0.1:57643 - "GET /v1/robots/1 HTTP/1.1" 200 OK

pathパラメーターの取得、DBからのクエリ取得、jsonレスポンスを返すことに成功していますね。
CURLでさっと書ける位ならいいのですが、実際開発では複雑なパラメーターがあったりで、APIのチェックはCURLだと面倒ですよね、ほかにもWebsocketだとWSCATだとか、コマンドラインで確認したりとやっている方もいるかと思いますが、オススメの拡張機能があるので是非使ってみてください。

Talend API Tester

Chorme拡張機能でTalend API Testerです。これがあればWebsocketもAPIも簡単にデバッグが簡単にできて、開発が楽になるとおもうのでおすすめです。最近はマイクロサービス化が進んできていて、アプリを作る場合、APIで作ることが増えてきているので、コマンドでデバッグがし辛いときは是非使ってみてください。

次回はバックグラウンドタスク実装

タスクキューのような待たずに実行させたい処理の追加として、参照回数をインクリメントさせる機能の実装について解説したいと思います。

今回の内容はこちらにコミットしています。