mirror of
https://github.com/correl/tornado-openapi3.git
synced 2024-11-25 03:00:14 +00:00
114 lines
5.1 KiB
Python
114 lines
5.1 KiB
Python
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.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):
|
|
@property
|
|
def spec(self) -> dict:
|
|
"""The OpenAPI 3 specification as a Python dictionary.
|
|
|
|
Override this in your request handlers to load or define your OpenAPI 3
|
|
spec.
|
|
|
|
"""
|
|
return dict()
|
|
|
|
@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(
|
|
create_spec(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()
|