Handling Incoming Requests

Tornado OpenAPI 3 allows you to validate requests coming in to your application against your OpenAPI 3.0 specification with little additional code.

Defining a base handler with your OpenAPI 3.0 specification

By extending OpenAPIRequestHandler, you can define your own base request handler with your specification attached, and use that for each of your specialized request handlers for your application.

import tornado.ioloop
import tornado.web
from tornado_openapi3.handler import OpenAPIRequestHandler


class MyRequestHandler(OpenAPIRequestHandler):
    spec_dict = {
        "openapi": "3.0.0",
        "info": {
            "title": "Simple Example",
            "version": "1.0.0",
        },
        "paths": {
            "/": {
                "get": {
                    "responses": {
                        "200": {
                            "description": "Index",
                            "content": {
                                "text/html": {
                                    "schema": {"type": "string"},
                                }
                            },
                        }
                    }
                }
            }
        },
    }


class RootHandler(MyRequestHandler):
    async def get(self):
        self.finish("Hello, World!")


if __name__ == "__main__":
    app = tornado.web.Application([(r"/", RootHandler)])
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

A more complex example

Your specification doesn’t need to be embedded in your code. You may wish to store it separately in your repository, or even templatize some aspects of it (like your application version). Doing so is as simple as overriding your request handler’s spec_dict property to load your specification however you see fit.

By default, the specification is compiled on every request. To achieve better performance, you may also wish to override the request handler’s spec_dict property to cache the result on your application object.

import logging
import pathlib

import tornado.ioloop
import tornado.web
from tornado_openapi3.handler import OpenAPIRequestHandler
import yaml

VERSION = "1.0.0"


class MyRequestHandler(OpenAPIRequestHandler):
    @property
    def spec_dict(self):
        return yaml.safe_load(self.render_string("openapi.yaml", version=VERSION))

    @property
    def spec(self):
        spec = getattr(self.application, "openapi_spec", None)
        if not spec:
            logging.info("Compiling OpenAPI spec")
            spec = super().spec
            setattr(self.application, "openapi_spec", spec)
        return spec


class RootHandler(MyRequestHandler):
    async def get(self):
        self.finish("Hello, World!")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    example_root = pathlib.Path(__file__).parent
    app = tornado.web.Application(
        [(r"/", RootHandler)], template_path=str(example_root / "templates")
    )
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

Adding custom deserializers

If your endpoints make use of content types beyond application/json, you must add them to this dictionary with a deserializing method that converts the raw body (as bytes or str) to Python objects.

import json

from tornado_openapi3.handler import OpenAPIRequestHandler


class ResourceHandler(OpenAPIRequestHandler):
    custom_media_type_deserializers = {
        "application/vnd.example.resource+json": json.loads,
    }

    ...

Adding custom formatters

If your schemas make use of format modifiers, you may specify them in this dictionary paired with a Formatter object that provides methods to validate values and unmarshal them into Python objects.

import datetime

from tornado_openapi3.handler import OpenAPIRequestHandler


class USDateFormatter:
    def validate(self, value: str) -> bool:
        return bool(re.match(r"^\d{1,2}/\d{1,2}/\d{4}$", value))

    def unmarshal(self, value: str) -> datetime.date:
        return datetime.datetime.strptime(value, "%m/%d/%Y").date()


class ResourceHandler(OpenAPIRequestHandler):
    custom_formatters = {
        "usdate": USDateFormatter(),
    }

    ...