TornadoでJSONを吐くAPI作るときにつかうことあれこれ

久しぶりにコード書くと、細かいところが思い出せない......

• ヘッダの値を読み取る

RequestHandlerのrequest.headersプロパティに、ヘッダ名がキーの辞書形式で保存されている。 例はX-HTTP-Method-Overrideの値を取得するところ。 もちろんヘッダが存在しないとValueError例外が出るので、Dictのget()メソッドなりで対処する。

method_override = self.request.headers['X-HTTP-Method-Override']

• 全HTTPメソッドの共通処理をつけくわえる。

RequestHandler.perpare() をオーバーライドして、処理を書く。
例えば、application/json以外のコンテンツを受け取る気がない場合は

class JSONAPIHandler(RequestHandler):

    def prepare(self):
        if self.request.headers.get('Content-Type') != 'application/json':
            raise HTTPError(406)

• JSONを受け取る

tornado.escape.json_decode()でbodyを辞書に変換すればOK。

import tornado.escape
# 中略

class JSONAPIHandler(RequestHandler):

    def post(self, *arg, **kwargs):
        request_json = json_decode(self.request.body)

• JSONを返す

tornado.escape.json_encode()でJSONにした文字列をRequestHandler.finish()やRequestHandler.write()でもいいが、 これら2つのメソッドに渡すオブジェクトが辞書の場合、内部でJSONエンコードしてくれるので便利。
ついでにContent-Typeヘッダの値をapplication/jsonに変更してれるため使えるなら使おう。

RequestHandler.write()のソースを引用

    def write(self, chunk):
        """
        pydocは略
        """
        if self._finished:
            raise RuntimeError("Cannot write() after finish()")
        if not isinstance(chunk, (bytes, unicode_type, dict)):
            message = "write() only accepts bytes, unicode, and dict objects"
            if isinstance(chunk, list):
                message += ". Lists not accepted for security reasons; see http://www.tornadoweb.org/en/stable/web.html#tornado.web.RequestHandler.write"
            raise TypeError(message)
        if isinstance(chunk, dict):
            chunk = escape.json_encode(chunk)
            self.set_header("Content-Type", "application/json; charset=UTF-8")
        chunk = utf8(chunk)
        self._write_buffer.append(chunk)

見てのとおり、JSONのオブジェクトでラップせずにリストを返信しようとすると、セキュリティ上の観点からダメと怒られるので注意。

• エラーもJSONにする

HTTPErrorや内部例外で5XXエラーを返すとき、デフォルトのままだとテキストがレスポンスされる。 JSONしか返す気のないAPIをつくる際には、ちょっと気持ち悪い RequestHandlerのwrite_error()メソッドをオーバーライドする。
自分は元のコードをベースにしてこんなのを使っている。

# 前略

    def write_error(self, status_code, **kwargs):

        res = OrderedDict([
            ('code', status_code),
            ('message', self._reason),
        ])

        if self.settings.get("serve_traceback") and "exc_info" in kwargs:
            res['exc_info'] = '\n'.join(traceback.format_exception(*kwargs["exc_info"]))

        self.finish(res)