diff --git a/tests/test_decorator_pydantic.py b/tests/test_decorator_pydantic.py new file mode 100644 index 0000000..abf5507 --- /dev/null +++ b/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(, +# {'/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" diff --git a/tornado_swagger/_builders.py b/tornado_swagger/_builders.py index b84660f..989e29d 100644 --- a/tornado_swagger/_builders.py +++ b/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", diff --git a/tornado_swagger/setup.py b/tornado_swagger/setup.py index 29bc845..e923e99 100644 --- a/tornado_swagger/setup.py +++ b/tornado_swagger/setup.py @@ -47,6 +47,7 @@ class SwaggerMethodInfo: request: typing.Optional[typing.Type[BaseModel]] = None query: typing.Optional[typing.Type[BaseModel]] = None tags: typing.Optional[typing.List[str]] = None + description: typing.Optional[str] = None def swagger_decorator( @@ -54,10 +55,16 @@ def swagger_decorator( responses: typing.Dict[int, typing.Dict[str, typing.Any]], request: typing.Optional[typing.Type[BaseModel]] = None, query: typing.Optional[typing.Type[BaseModel]] = None, - tags: typing.Optional[typing.List[str]] = None + tags: typing.Optional[typing.List[str]] = None, + description: typing.Optional[str] = None, ): def decorator(f: typing.Callable) -> typing.Callable: - f._swagger_info = SwaggerMethodInfo(responses, request, query, tags) + f._swagger_info = SwaggerMethodInfo( + responses=responses, + request=request, + query=query, + tags=tags, + description=description) return f return decorator