Добавить поддержку pydantic датаклассов

tests/test_decorator_pydantic.py

@@ -0,0 +1,254 @@
+import pytest
+import tornado.web
+
+from pydantic import BaseModel
+from pydantic.dataclasses import dataclass
+
+from tornado_swagger.const import API_OPENAPI_3_PYDANTIC
+from tornado_swagger.setup import swagger_decorator
+from tornado_swagger._builders import (
+    PydanticRoutesProcessor,
+    generate_doc_from_endpoints,
+    DEFAULT_SUCCESS_DESCRIPTION,
+    DEFAULT_FAIL_DESCRIPTION
+)
+
+SUCCESS_DESCRIPTION = "OK"
+NOT_FOUND_DESCRIPTION = "Not found"
+BASE_PATH = r"/api/items/(\d+)"
+DOC_PATH = "/api/items/{item_id}"
+
+class _ReqModel(BaseModel):
+    name: str
+    active: bool = True
+
+
+class _QueryModel(BaseModel):
+    limit: int
+    offset: int = 0
+
+
+@dataclass
+class _RespDC:
+    id: int
+    name: str
+
+
+class _DecoratedHandler(tornado.web.RequestHandler):
+    SUPPORTED_METHODS = ("POST", "PUT")
+
+    # заполнены все поля
+    @swagger_decorator(
+        request=_ReqModel,
+        query=_QueryModel,
+        description="Some test description",
+        responses={
+            200: {"model": _RespDC, "description": SUCCESS_DESCRIPTION},
+            404: {"description": NOT_FOUND_DESCRIPTION},
+        },
+        tags=["decorated"],
+    )
+    def post(self, item_id: int):
+        pass
+
+
+    # минимальное заполнение
+    @swagger_decorator(
+        responses={
+            200: {},
+            403: {}
+        },
+        tags=["decorated"],
+    )
+    def put(self, item_id: int):
+        pass
+
+# Пример распаршенного объекта
+# defaultdict(<class 'dict'>,
+#             {'/api/items/{item_id}': {'post': {'description': 'Some test '
+#                                                               'description',
+#                                                'parameters': [{'in': 'path',
+#                                                                'name': 'item_id',
+#                                                                'required': True,
+#                                                                'schema': {'format': 'int32',
+#                                                                           'type': 'integer'}},
+#                                                               {'in': 'query',
+#                                                                'name': 'limit',
+#                                                                'required': True,
+#                                                                'schema': {'title': 'Limit',
+#                                                                           'type': 'integer'}},
+#                                                               {'in': 'query',
+#                                                                'name': 'offset',
+#                                                                'required': False,
+#                                                                'schema': {'default': 0,
+#                                                                           'title': 'Offset',
+#                                                                           'type': 'integer'}}],
+#                                                'requestBody': {'content': {'application/json': {'schema': {'properties': {'active': {'default': True,
+#                                                                                                                                      'title': 'Active',
+#                                                                                                                                      'type': 'boolean'},
+#                                                                                                                           'name': {'title': 'Name',
+#                                                                                                                                    'type': 'string'}},
+#                                                                                                            'required': ['name'],
+#                                                                                                            'title': '_ReqModel',
+#                                                                                                            'type': 'object'}}},
+#                                                                'required': True},
+#                                                'responses': {200: {'content': {'application/json': {'schema': {'$ref': '#/components/schemas/_RespDC'}}},
+#                                                                    'description': 'OK'},
+#                                                              404: {'description': 'Not '
+#                                                                                   'found'}},
+#                                                'tags': ['decorated']},
+#                                       'put': {'parameters': [{'in': 'path',
+#                                                               'name': 'item_id',
+#                                                               'required': True,
+#                                                               'schema': {'format': 'int32',
+#                                                                          'type': 'integer'}}],
+#                                               'responses': {200: {'description': 'Successful '
+#                                                                                  'Response'},
+#                                                             403: {'description': 'Bad '
+#                                                                                  'request'}},
+#                                               'tags': ['decorated']}}})
+# ****************************************************************************************************
+# {'parameters': {},
+#  'schemas': {'_RespDC': {'properties': {'id': {'title': 'Id',
+#                                                'type': 'integer'},
+#                                         'name': {'title': 'Name',
+#                                                  'type': 'string'}},
+#                          'required': ['id', 'name'],
+#                          'title': '_RespDC',
+#                          'type': 'object'}}}
+
+
+
+
+def test_decorator_builds_paths_from_urlspec():
+    route = tornado.web.url(BASE_PATH, _DecoratedHandler)
+
+    routes = [route]
+    paths, components = PydanticRoutesProcessor().extract_paths_pydantic(routes)
+
+    # пути одинаковые т.к. аргументы пути совпадают, но методы разные
+    assert len(paths) == 1
+    assert  len(paths[DOC_PATH]) == 2
+
+def test_decorator_builds_paths_from_tuples():
+    route = (BASE_PATH, _DecoratedHandler)
+
+    routes = [route]
+    paths, components = PydanticRoutesProcessor().extract_paths_pydantic(routes)
+
+    # пути одинаковые т.к. аргументы пути совпадают, но методы разные
+    assert len(paths) == 1
+    assert  len(paths[DOC_PATH]) == 2
+
+
+def test_decorator_post():
+    route = tornado.web.url(BASE_PATH, _DecoratedHandler)
+
+    routes = [route]
+    paths, components = PydanticRoutesProcessor().extract_paths_pydantic(routes)
+
+    path = next((p for p in paths.keys()), None)
+    assert path == DOC_PATH
+    post_spec = next((params for params in paths.values()), {}).get("post")
+    assert post_spec
+
+    # Теги
+    assert "tags" in post_spec and post_spec["tags"] == ["decorated"]
+
+    # requestBody: схема инлайн, заголовок соответствует модели
+    req_schema = post_spec["requestBody"]["content"]["application/json"]["schema"]
+    assert req_schema["title"] == "_ReqModel"
+    assert req_schema["type"] == "object"
+    assert req_schema["required"] == ["name"]
+    # Параметры схемы
+    req_schema_properties = req_schema["properties"]
+    assert "name" in req_schema["properties"]
+    assert  req_schema_properties["name"]["type"] == "string"
+    assert "active" in req_schema["properties"]
+    assert  req_schema_properties["active"]["type"] == "boolean"
+    assert  req_schema_properties["active"]["default"] == True
+
+    # query-параметры из _QueryModel
+    qparams = {p["name"]: p for p in post_spec.get("parameters", []) if p["in"] == "query"}
+    assert "limit" in qparams and qparams["limit"]["schema"]["type"] == "integer"
+    assert qparams["limit"]["required"] is True
+    assert "offset" in qparams and qparams["offset"]["required"] is False
+    assert "limit" in qparams and qparams["offset"]["schema"]["type"] == "integer"
+
+    # path-параметр item_id (int)
+    pparams = [p for p in post_spec.get("parameters", []) if p["in"] == "path"]
+    assert any(p["name"] == "item_id" and p["schema"]["type"] in "integer" for p in pparams)
+
+    #ответы
+    resp_200 = post_spec["responses"][200]
+    ref = resp_200["content"]["application/json"]["schema"]["$ref"]
+    ref_name = ref.split("/")[-1]
+    assert ref_name == "_RespDC"
+    assert ref_name in components["schemas"]
+    assert resp_200["description"] == SUCCESS_DESCRIPTION
+
+    resp_404 = post_spec["responses"][404]
+    assert resp_404["description"] == NOT_FOUND_DESCRIPTION
+
+    # модель ответа (pydantic dataclass)
+    response = components["schemas"]["_RespDC"]
+    assert response["required"] == ["id", "name"]
+    assert response["properties"]["id"]["type"] == "integer"
+    assert response["properties"]["name"]["type"] == "string"
+
+
+def test_decorator_put():
+    # Декоратор с минимальным набором полей
+    route = tornado.web.url(BASE_PATH, _DecoratedHandler)
+
+    routes = [route]
+    paths, components = PydanticRoutesProcessor().extract_paths_pydantic(routes)
+
+    path = next((p for p in paths.keys()), None)
+    assert path == DOC_PATH
+    put_spec = next((params for params in paths.values()), {}).get("put")
+    assert put_spec
+
+    # Теги
+    assert "tags" in put_spec and put_spec["tags"] == ["decorated"]
+
+    # path-параметр item_id (int)
+    pparams = [p for p in put_spec.get("parameters", []) if p["in"] == "path"]
+    assert any(p["name"] == "item_id" and p["schema"]["type"] in "integer" for p in pparams)
+
+    # Описания кодов ответа
+    resp_200 = put_spec["responses"][200]
+    assert resp_200["description"] == DEFAULT_SUCCESS_DESCRIPTION
+
+    resp_403 = put_spec["responses"][403]
+    assert resp_403["description"] == DEFAULT_FAIL_DESCRIPTION
+
+
+def test_decorator_end_to_end_generate_doc_from_endpoints():
+    routes = [tornado.web.url(BASE_PATH, _DecoratedHandler)]
+
+    docs = generate_doc_from_endpoints(
+        routes=routes,
+        api_base_url="/",
+        description="",
+        api_version="1.0.0",
+        title="Test API",
+        contact="",
+        schemes=["http"],
+        security_definitions=None,
+        security=None,
+        api_definition_version=API_OPENAPI_3_PYDANTIC,
+    )
+
+    assert "openapi" in docs and docs["openapi"].startswith("3.")
+    assert DOC_PATH in docs["paths"]
+
+    post_spec = docs["paths"][DOC_PATH]["post"]
+    # базовые поля операции
+    assert "responses" in post_spec
+    assert "requestBody" in post_spec
+    assert "tags" in post_spec and post_spec["tags"] == ["decorated"]
+
+    # components содержит схему ответа
+    comp = docs["components"]["schemas"]
+    assert "_RespDC" in comp and comp["_RespDC"]["type"] == "object"

tornado_swagger/_builders.py

@@ -21,6 +21,9 @@
     os.path.join(os.path.dirname(__file__), "templates", "swagger.yaml")
 )
 SWAGGER_DOC_SEPARATOR = "---"
+DEFAULT_SUCCESS_DESCRIPTION = "Successful Response"
+DEFAULT_FAIL_DESCRIPTION = "Bad request"
+DEFAULT_INTERNAL_SERVER_ERROR_DESCRIPTION = "Internal Server Error"
 
 
 PYTHON_TO_OPENAPI_MAPPER = {
@@ -110,7 +113,13 @@ def __init__(self):
 
     def extract_paths_pydantic(self, routes):
         for route in routes:
-            tornado_route = tornado.web.url(*route)
+            if isinstance(route, tornado.web.URLSpec):
+                tornado_route = route
+            elif isinstance(route, (list, tuple)):
+                tornado_route = tornado.web.url(*route)
+            else:
+                raise TypeError(f"Unsupported route type: {type(route)!r}")
+
             for method_name, method_description in self._build_doc_from_pydantic_handler(
                     tornado_route.target
             ).items():
@@ -133,12 +142,13 @@ def _build_doc_from_pydantic_handler(self, handler):
                 response_models = swagger_info.responses
                 request_model = swagger_info.request
                 query_params = swagger_info.query
+                description = getattr(swagger_info, "description", None)
                 tags = swagger_info.tags
                 input_parameters = input_parameters_getter(method_callable)
                 out.update(
                     {
                         method_name: self.build_pydantic_docs(
-                            input_parameters, response_models, request_model, query_params, tags,
+                            input_parameters, response_models, request_model, query_params, tags, description=description
                         )
                     }
                 )
@@ -154,10 +164,10 @@ def _add_components_from_definitions(self, definitions: typing.Dict[str, typing.
     @staticmethod
     def _generate_default_description(status_code: int) -> str:
         if status_code < 400:
-            return "Successful Response"
+            return DEFAULT_SUCCESS_DESCRIPTION
         elif status_code < 500:
-            return "Bad request"
-        return "Internal Server Error"
+            return DEFAULT_FAIL_DESCRIPTION
+        return DEFAULT_INTERNAL_SERVER_ERROR_DESCRIPTION
 
     def build_pydantic_docs(
         self,
@@ -166,15 +176,19 @@ def build_pydantic_docs(
         request: typing.Optional[typing.Type[BaseModel]] = None,
         query: typing.Optional[typing.Type[BaseModel]] = None,
         tags: typing.Optional[typing.List[str]] = None,
+        *,
+        description=None,
     ):
         result = {}
 
+        if description:
+            result["description"] = description
         parameters = self._build_input_and_query_doc(input_parameters, query)
         if parameters:
             result["parameters"] = parameters
 
         if request:
-            model_spec = request.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+            model_spec = self.get_pydantic_schema(request)
             if "definitions" in model_spec:
                 self._add_components_from_definitions(model_spec.pop("definitions"))
 
@@ -189,11 +203,17 @@ def build_pydantic_docs(
 
         responses = {}
         for status_code, response_model in response_models.items():
-            model = response_model["model"]
+            model = response_model.get("model", None)
             description = response_model.get("description", None)
             if not description:
                 description = self._generate_default_description(status_code)
-            model_spec = model.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+
+            # если ручка отвечает только кодом, без модели
+            if model is None:
+                responses[status_code] = {"description": description}
+                continue
+
+            model_spec = self.get_pydantic_schema(model)
             model_name = model.__name__
             # could cause conflicts for classes with same name
             if model_name not in self.components["schemas"]:
@@ -218,8 +238,20 @@ def build_pydantic_docs(
         return result
 
     @staticmethod
-    def _build_request_body_doc(model: BaseModel) -> dict:
-        model_schema = model.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+    def get_pydantic_schema(model) -> dict:
+        # если BaseModel - можем вытащить напрямую
+        if hasattr(model, "schema"):
+            return model.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+        # если датакласс (pydantic 1.1) - тащим через встроенную модель
+        # TODO в 2.0 интерфейс поменялся, нужно будет доработать
+        if hasattr(model, "__pydantic_model__"):
+            return model.__pydantic_model__.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+
+        raise TypeError(f"Unsupported model type for OpenAPI schema: {model}")
+
+
+    def _build_request_body_doc(self, model: BaseModel) -> dict:
+        model_schema = self.get_pydantic_schema(model)
 
         request_body = {
             "content": {
@@ -249,7 +281,7 @@ def _build_input_and_query_doc(
                 })
 
         if query:
-            query_schema = query.schema(by_alias=False, ref_template="#/components/schemas/{model}")
+            query_schema = PydanticRoutesProcessor.get_pydantic_schema(query)
             for parameter_name, schema in query_schema["properties"].items():
                 parameters.append({
                     "in": "query",