tornado-openapi3/tornado_openapi3/handler.py

import asyncio
import logging

from openapi_core import create_spec  # type: ignore
from openapi_core.exceptions import OpenAPIError  # type: ignore
from openapi_core.deserializing.exceptions import DeserializeError  # type: ignore
from openapi_core.schema.specs.models import Spec  # type: ignore
from openapi_core.schema.media_types.exceptions import (  # type: ignore
    InvalidContentType,
)
from openapi_core.templating.paths.exceptions import (  # type: ignore
    OperationNotFound,
    PathNotFound,
)
from openapi_core.unmarshalling.schemas.exceptions import ValidateError  # type: ignore
from openapi_core.validation.exceptions import InvalidSecurity  # type: ignore
import tornado.web  # type: ignore

from tornado_openapi3.requests import RequestValidator

logger = logging.getLogger(__name__)


class OpenAPIRequestHandler(tornado.web.RequestHandler):
    """Base class for HTTP request handlers.

    A request handler extending :py:class:`tornado.web.RequestHandler` providing
    OpenAPI spec validation on incoming requests and translating errors into
    appropriate HTTP responses.

    """

    @property
    def spec_dict(self) -> dict:
        """The OpenAPI 3 specification

        Override this in your request handlers to load or define your OpenAPI 3
        spec.

        :rtype: dict

        """
        raise NotImplementedError()

    @property
    def spec(self) -> Spec:
        """The OpenAPI 3 specification.

        Override this in your request handlers to customize how your OpenAPI 3
        spec is loaded and validated.

        :rtype: :py:class:`openapi_core.schema.specs.model.Spec`

        """
        return create_spec(self.spec_dict, validate_spec=False)

    @property
    def custom_media_type_deserializers(self) -> dict:
        """A dictionary mapping media types to deserializing functions.

        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.

        """
        return dict()

    async def prepare(self) -> None:
        """Called at the beginning of a request before *get/post/etc*.

        Performs OpenAPI validation of the incoming request. Problems
        encountered while validating the request are translated to HTTP error
        codes:

        +--------------------------+----------+----------------------------------------+
        |OpenAPI Errors            |Error Code|Description                             |
        +--------------------------+----------+----------------------------------------+
        |``PathNotFound``          |``404``   |Could not find the path for this request|
        |                          |          |in the OpenAPI specification.           |
        +--------------------------+----------+----------------------------------------+
        |``OperationNotFound``     |``405``   |Could not find the operation specified  |
        |                          |          |for this request in the OpenAPI         |
        |                          |          |specification.                          |
        +--------------------------+----------+----------------------------------------+
        |``DeserializeError``,     |``400``   |The message body could not be decoded or|
        |``ValidateError``         |          |did not validate against the specified  |
        |                          |          |schema.                                 |
        +--------------------------+----------+----------------------------------------+
        |``InvalidSecurity``       |``401``   |Required authorization was missing from |
        |                          |          |the request.                            |
        +--------------------------+----------+----------------------------------------+
        |``InvalidContentType``    |``415``   |The content type of the request did not |
        |                          |          |match any of the types in the OpenAPI   |
        |                          |          |specification.                          |
        +--------------------------+----------+----------------------------------------+
        |Any other ``OpenAPIError``|``500``   |An unexpected error occurred.           |
        +--------------------------+----------+----------------------------------------+

        To provide content in these error requests, you may override
        :meth:`on_openapi_error`.

        """
        maybe_coro = super().prepare()
        if maybe_coro and asyncio.iscoroutine(maybe_coro):  # pragma: no cover
            await maybe_coro

        validator = RequestValidator(
            self.spec,
            custom_media_type_deserializers=self.custom_media_type_deserializers,
        )
        result = validator.validate(self.request)
        try:
            result.raise_for_errors()
        except PathNotFound as e:
            self.on_openapi_error(404, e)
        except OperationNotFound as e:
            self.on_openapi_error(405, e)
        except (DeserializeError, ValidateError) as e:
            self.on_openapi_error(400, e)
        except InvalidSecurity as e:
            self.on_openapi_error(401, e)
        except InvalidContentType as e:
            self.on_openapi_error(415, e)
        except OpenAPIError as e:
            logger.exception("Unexpected validation failure")
            self.on_openapi_error(500, e)
        self.validated = result

    def on_openapi_error(self, status_code: int, error: OpenAPIError) -> None:
        """Sets an HTTP status code and finishes the request.

        By default, no content is returned. To provide more informative
        responses, you may override this method.

        """
        self.set_status(status_code)
        self.finish()