Responder使ってみた際のポイント・注意点（Pytestも書いてみたよ）

17674 ワード
Python3 responder Python Python テキストリンク

はじめに

最近流行っている（？） Pythonの新しい WebフレームワークのResponderを使って、ちょっとしたAPIを作ってみたところ、少し詰まったところがあったので書きます。
かなり高機能ながら、使っている事例が少なくて参考になるコードも少ないので少しでも参考になればと思います。

またPytestの事例も少なかったのでご参考になればと思います。

基本編

まずは基本編
ここらへんは結構記事がありますがかゆいところに手が届かない気もするのでもう少し書いてみます。

Route, Background tasks, JSON response

Responderの特徴の一つとして非同期のAPIを簡単に実装できることがありますが、本当に簡単です。
下のように書くだけで簡単に非同期なAPIを作れます。

ここらへんは公式のQuick Start!にだいたい書いてあります。

slow_job.py

import responder

api = responder.API()

@api.route("/data/upload") # 1.デコレータで簡単route、Flaskと同じですね
async def call_slow_job(req, resp): # 2.request, responseの順でパラメータに
    request = await req.media() # ★注意1
    slow_job(request)
    resp.media = {"status": "ok"} # 4.request, responseも.mediaでDictとしてJSONを扱える


@api.background.task # 3.非同期で処理させたい関数
def slow_job(request):
    # 重たい処理
    sleep(30)
    return True

ポイント

  1. エンドポイントになる関数にroute デコレータをつけることで簡単に作れます。 ここらへんはFlaskと同じですね
  2. request, responseは関数のパラメータとして設定されます
  3. background.taskを利用することで、非同期なAPIが簡単に作れます。重たいバッチ処理を受け付けるようなAPIに最適です
  4. requestもresponseも.media()でDict型としてJSONを扱えます

注意1

Background taskを使った場合、エンドポイントになる関数内の処理は意図せず非同期で処理されてしまうみたい？
関数の戻り値を後続の処理で使いたい場合、async で非同期関数を宣言し、await で処理を待ちましょう。

MarshmallowでValidationしてError Responseを返す

marshmallowを使って定義したSchemaに沿ったRequestか検証することができます
また、その際のハンドリング方法を紹介します

どちらかというと、Responderというよりmarshmallowの解説っぽい・・・

公式ではSchema/Validationではなく、OpenAPIの説明でちらっと載ってます
OpenAPI Schema Support

schema_example.py
from marshmallow import Schema, fields, ValidationError

@api.schema("DownloadReq")
class DownloadReqSchema(Schema):
    uploadId = fields.Str(required=True) # 1. Schema定義


@api.schema("ErrorResp")
class ErrorRespSchema(Schema):
    error = fields.Str()
    errorDate = fields.Date()


class ErrorModel:
    def __init__(self, error):
        self.error = str(error)
        self.errorDate = datetime.datetime.now()


@api.route("/convert/pdf/download")
async def download_result(req, resp):
    request = await req.media()
    try:
        data = DownloadReqSchema(strict=True).load(request).data # 2. RequestをSchemaで検証

    except ValidationError as error: # 3. VaridationErrorが返る
        resp.status_code = api.status_codes.HTTP_400
        resp.media = ErrorRespSchema().dump(ErrorModel(error)).data # 4. ResponseもSchema定義可能

        return # 5. Returnすればその時responseに設定されているものがresponse
    with open(os.path.join(upload_id, "result.pdf"), "rb") as result_pdf:
        resp.headers["Content-Type"] = "application/pdf"
        resp.content = result_pdf.read() # 6. おまけ ファイルをダウンロードする方法

ポイント

  1. marshmallowのSchema, fieldsを使ってSchemaを定義します required=Trueをfieldsに設定することでvalidationの対象になります
  2. requestを1で設定したSchemaでロードすることでDeserializingとValidateをすることができます strict=Trueを設定しないとvalidationErrorがでません
  3. validateに失敗するとVaridationErrorが返るのでexceptします
  4. responseもschemaでserializeできます　あらかじめモデルを作ってそこに格納した後dumpします　ここではErrorModelクラスを作っていますがnamedtupleで簡単に作ることもできます
  5. except節でAPI的にErrorを返したい場合はresponseに設定後、関数をreturnすればその時設定されている値が返却されます
  6. おまけ: response.contentにbyteモードで開いたファイルを読み込ませることでファイルダウンロードを実現できます

応用編(Pytest)

Pytestでどう調理するかあまり記事がなかったのでまとめます。

HTTPリクエストを疑似

Pytestのfixtureを使えば簡単にテストできます。

test_slow_job.py
import json
from unittest.mock import patch

import pytest
import hogehoge as target

@pytest.fixture # 1. fixtureを設定
def api():
    return target.api 

def test_call_slow_job (api):
    event = json.dumps({"fileName": "test.pdf",
                        "contentType": "image/png"})
    with patch.object(target, "slow_job", return_value=True) as mock_heavy_job: # Unit Test的にBackground TaskはMock化
        r = api.requests.post("/data/upload", event) # 2. 実際にコールしてresponseを取得
        assert r.status_code == 200
        json_response = json.loads(r.text)
        assert "id" in json_response
        assert "release_date" in json_response
        assert json_response["id"] == "hoge"

ポイント

  1. pytest.fixture であらかじめResponderAPIをインスタンス化しておきます　ここらへんは公式Docにも書いてありますので、普通によくやる方法らしい Building and Testing with Responder
  2. get methodはrequests.get post methodはrequests.postでコールします　第一引数にURI, 第二引数にpayloadを入れます

Background TaskをTest

Background Taskを設定した関数へのTestはBackground Taskの挙動からFutureオブジェクトが返ります。
Source
下記のようにテストしてあげればよいです。

test_slow_job.py
def test_slow_job(tmpdir):
    future = target.heavy_job(["hoge"], tmpdir) # 1. Background Taskの戻りはFutureが返ります
    actual = future.result() # 2. result()で結果を取得　★注意2
    expected = True
    assert actual == expected

ポイント

  1. Background Taskでデコレートした関数の戻りはFutureインスタンスになります
  2. futureインスタンスに対してresult()メソッドをコールすることで、結果が取得できます

注意2

IDEとかでLinterを使っているとデコレータの挙動なんてお構いなしなので、関数の戻り値にresult()なんてメソッドないですよと怒られる。

おわりに

ResponderはほかにもOpenAPIやWebsocketやgrapheneなどできることは山ほどありそうです。素晴らしいですね。

今回勉強したことを使ってちょっとしたAPIを作ってみました。
今回紹介したこと以外にもOpenAPIやCORSとかも使ってみました。

Ebook_homebrew　Github

もう少し色々触ってみようかと思います。

おまけ

★Herokuにもデプロイしております https://ebook-homebrew.herokuapp.com/docs

★Vue.jsを使ったフロントも拙作ながら作ってみました Source Heroku

※ 特にHerokuのスケールダウンの細工をしていないので、立ち上がりが悪いです。

参考

公式Doc

Python responder 入門のために… 下調べ

marshmallow公式Doc

[JS]関数第1章
プロジェクトGitHubにアップロード