More than 5 years have passed since last update.

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

Last updated at 2019-05-05Posted at 2019-05-05

はじめに

最近流行っている（？） 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

ポイント

エンドポイントになる関数にroute デコレータをつけることで簡単に作れます。ここらへんはFlaskと同じですね
request, responseは関数のパラメータとして設定されます
background.taskを利用することで、非同期なAPIが簡単に作れます。重たいバッチ処理を受け付けるようなAPIに最適です
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. おまけ ファイルをダウンロードする方法

ポイント

marshmallowのSchema, fieldsを使ってSchemaを定義します required=Trueをfieldsに設定することでvalidationの対象になります
requestを1で設定したSchemaでロードすることでDeserializingとValidateをすることができます strict=Trueを設定しないとvalidationErrorがでません
validateに失敗するとVaridationErrorが返るのでexceptします
responseもschemaでserializeできます　あらかじめモデルを作ってそこに格納した後dumpします　ここではErrorModelクラスを作っていますがnamedtupleで簡単に作ることもできます
except節でAPI的にErrorを返したい場合はresponseに設定後、関数をreturnすればその時設定されている値が返却されます
おまけ: 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"

ポイント

pytest.fixture であらかじめResponderAPIをインスタンス化しておきます　ここらへんは公式Docにも書いてありますので、普通によくやる方法らしい Building and Testing with Responder
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