Initial commit

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+max-line-length = 88
+extend-ignore = E203

.gitignore

@@ -0,0 +1,12 @@
+*~
+.DS_Store
+.idea
+poetry.lock
+.python-version
+*.log
+tmp/
+
+*.py[cod]
+*.egg
+build
+htmlcov

openapi3/__init__.py

@@ -0,0 +1,3 @@
+from openapi3.requests import TornadoRequestFactory
+
+__all__ = ["TornadoRequestFactory"]

openapi3/requests.py

@@ -0,0 +1,34 @@
+import itertools
+
+from openapi_core.validation.request.datatypes import (  # type: ignore
+    RequestParameters,
+    OpenAPIRequest,
+)
+from tornado.httputil import HTTPServerRequest
+from werkzeug.datastructures import ImmutableMultiDict, Headers
+
+
+class TornadoRequestFactory:
+    @classmethod
+    def create(cls, request: HTTPServerRequest) -> OpenAPIRequest:
+        if request.uri:
+            path, _, _ = request.uri.partition("?")
+        else:
+            path = ""
+        query_arguments: ImmutableMultiDict[str, str] = ImmutableMultiDict(
+            itertools.chain(
+                *[
+                    [(k, v.decode("utf-8")) for v in vs]
+                    for k, vs in request.query_arguments.items()
+                ]
+            )
+        )
+        return OpenAPIRequest(
+            full_url_pattern=path,
+            method=request.method.lower() if request.method else "get",
+            parameters=RequestParameters(
+                query=query_arguments, header=Headers(request.headers)
+            ),
+            body=request.body.decode("utf-8"),
+            mimetype="text/html",
+        )

pyproject.toml

@@ -0,0 +1,27 @@
+[tool.poetry]
+name = "tornado-openapi3"
+version = "0.1.0"
+description = "Tornado OpenAPI 3 request and response validation"
+authors = ["Correl Roush <correl@gmail.com>"]
+license = "MIT"
+
+[tool.poetry.dependencies]
+python = "^3.7"
+tornado = "^6"
+openapi-core = "^0.13.4"
+
+[tool.poetry.dev-dependencies]
+black = "^20.8b1"
+mypy = "^0.782"
+hypothesis = "^5.35.3"
+flake8 = "^3.8.3"
+pytest = "^6.0.2"
+pytest-flake8 = "^1.0.6"
+pytest-mypy = "^0.7.0"
+
+[tool.pytest.ini_options]
+addopts = "--flake8 --mypy"
+
+[build-system]
+requires = ["poetry>=0.12"]
+build-backend = "poetry.masonry.api"

tests/test_requests.py

@@ -0,0 +1,27 @@
+import unittest
+
+import attr
+from openapi_core.validation.request.datatypes import (  # type: ignore
+    RequestParameters,
+    OpenAPIRequest,
+)
+from tornado.httputil import HTTPServerRequest
+from werkzeug.datastructures import ImmutableMultiDict
+
+from openapi3 import TornadoRequestFactory
+
+
+class TestRequestFactory(unittest.TestCase):
+    def test_request(self) -> None:
+        tornado_request = HTTPServerRequest(
+            method="GET", uri="http://example.com/foo?bar=baz"
+        )
+        expected = OpenAPIRequest(
+            full_url_pattern="http://example.com/foo",
+            method="get",
+            parameters=RequestParameters(query=ImmutableMultiDict([("bar", "baz")])),
+            body="",
+            mimetype="text/html",
+        )
+        openapi_request = TornadoRequestFactory.create(tornado_request)
+        self.assertEqual(attr.asdict(expected), attr.asdict(openapi_request))