From 84a300ef842bc41b4955973a280366952097fd32 Mon Sep 17 00:00:00 2001 From: Mohammed Date: Fri, 22 Mar 2019 22:54:48 +0300 Subject: [PATCH] Formatting according to guide --- fastapi/applications.py | 324 ++++++++++++------------ fastapi/openapi/models.py | 15 +- fastapi/openapi/utils.py | 20 +- fastapi/routing.py | 387 +++++++++++++++-------------- fastapi/utils.py | 4 +- tests/test_additional_responses.py | 373 +++++++++------------------ 6 files changed, 489 insertions(+), 634 deletions(-) diff --git a/fastapi/applications.py b/fastapi/applications.py index 5ead7db91..db0c85169 100644 --- a/fastapi/applications.py +++ b/fastapi/applications.py @@ -2,8 +2,8 @@ from typing import Any, Callable, Dict, List, Optional, Type from fastapi import routing from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html -from fastapi.openapi.utils import get_openapi from fastapi.openapi.models import AdditionalResponse, AdditionalResponseDescription +from fastapi.openapi.utils import get_openapi from pydantic import BaseModel from starlette.applications import Starlette from starlette.exceptions import ExceptionMiddleware, HTTPException @@ -105,23 +105,23 @@ class FastAPI(Starlette): self.add_exception_handler(HTTPException, http_exception) def add_api_route( - self, - path: str, - endpoint: Callable, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - methods: List[str] = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + endpoint: Callable, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + methods: List[str] = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> None: self.router.add_api_route( path, @@ -142,22 +142,22 @@ class FastAPI(Starlette): ) def api_route( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - methods: List[str] = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + methods: List[str] = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: def decorator(func: Callable) -> Callable: self.router.add_api_route( @@ -182,31 +182,33 @@ class FastAPI(Starlette): return decorator def include_router( - self, - router: routing.APIRouter, - *, - prefix: str = "", - tags: List[str] = None, - additional_responses: List[AdditionalResponse] = [], + self, + router: routing.APIRouter, + *, + prefix: str = "", + tags: List[str] = None, + additional_responses: List[AdditionalResponse] = [], ) -> None: - self.router.include_router(router, prefix=prefix, tags=tags, additional_responses=additional_responses,) + self.router.include_router( + router, prefix=prefix, tags=tags, additional_responses=additional_responses + ) def get( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.get( path, @@ -225,21 +227,21 @@ class FastAPI(Starlette): ) def put( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.put( path, @@ -258,21 +260,21 @@ class FastAPI(Starlette): ) def post( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.post( path, @@ -291,21 +293,21 @@ class FastAPI(Starlette): ) def delete( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.delete( path, @@ -324,21 +326,21 @@ class FastAPI(Starlette): ) def options( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.options( path, @@ -357,21 +359,21 @@ class FastAPI(Starlette): ) def head( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.head( path, @@ -390,21 +392,21 @@ class FastAPI(Starlette): ) def patch( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.patch( path, @@ -423,21 +425,21 @@ class FastAPI(Starlette): ) def trace( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.router.trace( path, diff --git a/fastapi/openapi/models.py b/fastapi/openapi/models.py index e94e087fa..70865548f 100644 --- a/fastapi/openapi/models.py +++ b/fastapi/openapi/models.py @@ -1,10 +1,10 @@ import logging from enum import Enum -from typing import Any, Dict, List, Optional, Union, Type, ClassVar, Callable +from typing import Any, Callable, ClassVar, Dict, List, Optional, Type, Union from pydantic import BaseModel, Schema as PSchema -from pydantic.types import UrlStr from pydantic.fields import Field +from pydantic.types import UrlStr try: import email_validator @@ -351,19 +351,12 @@ class BaseAdditionalResponse(BaseModel): class AdditionalResponse(BaseAdditionalResponse): status_code: int = PSchema( - ..., - ge=100, - le=540, - title='Status Code', - description='HTTP status code', + ..., ge=100, le=540, title="Status Code", description="HTTP status code" ) # NOTE: waiting for pydantic to allow `typing.Type[BasicModel]` type # so, going for `Any` and extra validation on # routing methods - models: Optional[List[Any]] = PSchema( - [], - title='Additional Response Models', - ) + models: Optional[List[Any]] = PSchema([], title="Additional Response Models") class AdditionalResponseDescription(BaseAdditionalResponse): diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py index 024fb50c8..a1271676f 100644 --- a/fastapi/openapi/utils.py +++ b/fastapi/openapi/utils.py @@ -207,24 +207,22 @@ def get_openapi_path( } for add_response_code, add_response in route.additional_responses.items(): add_response_schema = {} - if (add_response.content_type or route.content_type.media_type - ) == 'application/json' and add_response.schema_field is not None: + if ( + add_response.content_type or route.content_type.media_type + ) == "application/json" and add_response.schema_field is not None: add_response_schema, _ = field_schema( add_response.schema_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX, ) add_content = { - add_response.content_type or - route.content_type.media_type: { - "schema": add_response_schema, - }, + add_response.content_type + or route.content_type.media_type: {"schema": add_response_schema} + } + operation["responses"][str(add_response_code)] = { + "description": add_response.description, + "content": add_content, } - operation["responses"][str(add_response_code)] = \ - { - "description": add_response.description, - "content": add_content, - } path[method.lower()] = operation return path, security_schemes, definitions diff --git a/fastapi/routing.py b/fastapi/routing.py index 57bbe7a63..d7ef1c7f7 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -1,14 +1,14 @@ import asyncio import inspect import logging -from typing import Any, Callable, List, Optional, Type, Dict, Union +from typing import Any, Callable, Dict, List, Optional, Type, Union from fastapi import params from fastapi.dependencies.models import Dependant from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies from fastapi.encoders import jsonable_encoder -from fastapi.utils import UnconstrainedConfig from fastapi.openapi.models import AdditionalResponse, AdditionalResponseDescription +from fastapi.utils import UnconstrainedConfig from pydantic import BaseModel, Schema from pydantic.error_wrappers import ErrorWrapper, ValidationError from pydantic.fields import Field @@ -146,31 +146,30 @@ class APIRoute(routing.Route): for add_response in additional_responses: if isinstance(add_response, int): continue - assert add_response.status_code not in existed_codes, f"(Duplicated Status Code): Response with status code [{add_response.status_code}] already defined!" + assert ( + add_response.status_code not in existed_codes + ), f"(Duplicated Status Code): Response with status code [{add_response.status_code}] already defined!" existed_codes.append(add_response.status_code) - response_models = [ - m for m in\ - add_response.models - ] + response_models = [m for m in add_response.models] valid_response_models = True try: - valid_response_models = all([ - issubclass(m, BaseModel) - for m in response_models - ]) + valid_response_models = all( + [issubclass(m, BaseModel) for m in response_models] + ) except TypeError as te: valid_response_models = False if not valid_response_models: raise ValueError( "All response models must be " "a subclass of `pydantic.BaseModel` " - "model.", + "model." ) - if (add_response.content_type == 'application/json' or lenient_issubclass( - content_type, JSONResponse)): + if add_response.content_type == "application/json" or lenient_issubclass( + content_type, JSONResponse + ): if len(response_models): schema_field = Field( - name=f'Additional_response_{add_response.status_code}', + name=f"Additional_response_{add_response.status_code}", type_=Union[tuple(response_models)], class_validators=[], default=None, @@ -186,9 +185,8 @@ class APIRoute(routing.Route): description=add_response.description, content_type=add_response.content_type, schema_field=schema_field, - ) - self.additional_responses[add_response.status_code] = \ - add_resp_description + ) + self.additional_responses[add_response.status_code] = add_resp_description self.deprecated = deprecated if methods is None: methods = ["GET"] @@ -216,23 +214,23 @@ class APIRoute(routing.Route): class APIRouter(routing.Router): def add_api_route( - self, - path: str, - endpoint: Callable, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - methods: List[str] = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + endpoint: Callable, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + methods: List[str] = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> None: route = APIRoute( path, @@ -254,22 +252,22 @@ class APIRouter(routing.Router): self.routes.append(route) def api_route( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - methods: List[str] = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + methods: List[str] = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: def decorator(func: Callable) -> Callable: self.add_api_route( @@ -294,12 +292,12 @@ class APIRouter(routing.Router): return decorator def include_router( - self, - router: "APIRouter", - *, - prefix: str = "", - tags: List[str] = None, - additional_responses: List[AdditionalResponse] = [], + self, + router: "APIRouter", + *, + prefix: str = "", + tags: List[str] = None, + additional_responses: List[AdditionalResponse] = [], ) -> None: if prefix: assert prefix.startswith("/"), "A path prefix must start with '/'" @@ -310,20 +308,20 @@ class APIRouter(routing.Router): if isinstance(route, APIRoute): # really ugly hack and repitition prev_add_resp = route.additional_responses - existed_codes = [422, route.status_code - ] + [int(c) for c in prev_add_resp.keys()] + existed_codes = [422, route.status_code] + [ + int(c) for c in prev_add_resp.keys() + ] for add_response in additional_responses: - assert add_response.status_code not in existed_codes, f"(Duplicated Status Code): Response with status code [{add_response.status_code}] already defined!" + assert ( + add_response.status_code not in existed_codes + ), f"(Duplicated Status Code): Response with status code [{add_response.status_code}] already defined!" existed_codes.append(add_response.status_code) - response_models = [ - m for m in\ - add_response.models - ] + response_models = [m for m in add_response.models] valid_response_models = True try: - valid_response_models = all([ - issubclass(m, BaseModel) for m in response_models - ]) + valid_response_models = all( + [issubclass(m, BaseModel) for m in response_models] + ) except AttributeError as ae: valid_response_models = False if not valid_response_models: @@ -332,11 +330,13 @@ class APIRouter(routing.Router): "a subclass of `pydantic.BaseModel`" "model." ) - if (add_response.content_type == 'application/json' or lenient_issubclass( - route.content_type, JSONResponse)): + if ( + add_response.content_type == "application/json" + or lenient_issubclass(route.content_type, JSONResponse) + ): if len(response_models): schema_field = Field( - name=f'Additional_response_{add_response.status_code}', + name=f"Additional_response_{add_response.status_code}", type_=Union[tuple(response_models)], class_validators=[], default=None, @@ -352,9 +352,10 @@ class APIRouter(routing.Router): description=add_response.description, content_type=add_response.content_type, schema_field=schema_field, - ) - route.additional_responses[add_response.status_code] = \ - add_resp_description + ) + route.additional_responses[ + add_response.status_code + ] = add_resp_description self.add_api_route( prefix + route.path, route.endpoint, @@ -382,21 +383,21 @@ class APIRouter(routing.Router): ) def get( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -416,21 +417,21 @@ class APIRouter(routing.Router): ) def put( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -450,21 +451,21 @@ class APIRouter(routing.Router): ) def post( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -484,21 +485,21 @@ class APIRouter(routing.Router): ) def delete( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -518,21 +519,21 @@ class APIRouter(routing.Router): ) def options( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -552,21 +553,21 @@ class APIRouter(routing.Router): ) def head( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -586,21 +587,21 @@ class APIRouter(routing.Router): ) def patch( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, @@ -620,21 +621,21 @@ class APIRouter(routing.Router): ) def trace( - self, - path: str, - *, - response_model: Type[BaseModel] = None, - status_code: int = 200, - tags: List[str] = None, - summary: str = None, - description: str = None, - response_description: str = "Successful Response", - additional_responses: List[AdditionalResponse] = [], - deprecated: bool = None, - operation_id: str = None, - include_in_schema: bool = True, - content_type: Type[Response] = JSONResponse, - name: str = None, + self, + path: str, + *, + response_model: Type[BaseModel] = None, + status_code: int = 200, + tags: List[str] = None, + summary: str = None, + description: str = None, + response_description: str = "Successful Response", + additional_responses: List[AdditionalResponse] = [], + deprecated: bool = None, + operation_id: str = None, + include_in_schema: bool = True, + content_type: Type[Response] = JSONResponse, + name: str = None, ) -> Callable: return self.api_route( path=path, diff --git a/fastapi/utils.py b/fastapi/utils.py index aa25cfabe..d0ffbb5d8 100644 --- a/fastapi/utils.py +++ b/fastapi/utils.py @@ -33,9 +33,7 @@ def get_flat_models_from_routes( if route.additional_responses: for _, add_response in route.additional_responses.items(): if add_response.schema_field is not None: - responses_from_routes.append( - add_response.schema_field, - ) + responses_from_routes.append(add_response.schema_field) flat_models = get_flat_models_from_fields( body_fields_from_routes + responses_from_routes ) diff --git a/tests/test_additional_responses.py b/tests/test_additional_responses.py index b7bfae224..ddd5b7664 100644 --- a/tests/test_additional_responses.py +++ b/tests/test_additional_responses.py @@ -13,28 +13,22 @@ class Item(BaseModel): class Response400(BaseModel): - '''HTTP 4xx Response Schema''' + """HTTP 4xx Response Schema""" + title: str detail: str - error_code: int # functional error ref + error_code: int # functional error ref response_403 = AdditionalResponse( - status_code = 403, - description = 'Forbidden', - models = [ - Response400, - ], + status_code=403, description="Forbidden", models=[Response400] ) -additional_responses = [ - response_403, -] +additional_responses = [response_403] + @app.api_route( - "/items/{item_id}", - methods=["GET"], - additional_responses=additional_responses, + "/items/{item_id}", methods=["GET"], additional_responses=additional_responses ) def get_items(item_id: str): return {"item_id": item_id} @@ -51,42 +45,27 @@ app.add_api_route( ) -@app.delete( - "/items/{item_id}", - additional_responses=additional_responses, -) +@app.delete("/items/{item_id}", additional_responses=additional_responses) def delete_item(item_id: str, item: Item): return {"item_id": item_id, "item": item} -@app.head( - "/items/{item_id}", - additional_responses=additional_responses, -) +@app.head("/items/{item_id}", additional_responses=additional_responses) def head_item(item_id: str): return JSONResponse(headers={"x-fastapi-item-id": item_id}) -@app.options( - "/items/{item_id}", - additional_responses=additional_responses, -) +@app.options("/items/{item_id}", additional_responses=additional_responses) def options_item(item_id: str): return JSONResponse(headers={"x-fastapi-item-id": item_id}) -@app.patch( - "/items/{item_id}", - additional_responses=additional_responses, -) +@app.patch("/items/{item_id}", additional_responses=additional_responses) def patch_item(item_id: str, item: Item): return {"item_id": item_id, "item": item} -@app.trace( - "/items/{item_id}", - additional_responses=additional_responses, -) +@app.trace("/items/{item_id}", additional_responses=additional_responses) def trace_item(item_id: str): return JSONResponse(media_type="message/http") @@ -95,29 +74,20 @@ client = TestClient(app) openapi_schema = { "openapi": "3.0.2", - "info": { - "title": "Fast API", - "version": "0.1.0" - }, + "info": {"title": "Fast API", "version": "0.1.0"}, "paths": { "/items/{item_id}": { "get": { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -126,44 +96,34 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Get Items Get", - "operationId": - "get_items_items__item_id__get", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Get Items Get", + "operationId": "get_items_items__item_id__get", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], }, "delete": { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -172,32 +132,26 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Delete Item Delete", - "operationId": - "delete_item_items__item_id__delete", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Delete Item Delete", + "operationId": "delete_item_items__item_id__delete", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], "requestBody": { "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Item" - } + "schema": {"$ref": "#/components/schemas/Item"} } }, "required": True, @@ -207,19 +161,13 @@ openapi_schema = { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -227,9 +175,7 @@ openapi_schema = { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -238,44 +184,34 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Options Item Options", - "operationId": - "options_item_items__item_id__options", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Options Item Options", + "operationId": "options_item_items__item_id__options", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], }, "head": { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -284,44 +220,34 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Head Item Head", - "operationId": - "head_item_items__item_id__head", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Head Item Head", + "operationId": "head_item_items__item_id__head", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], }, "patch": { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -330,32 +256,26 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Patch Item Patch", - "operationId": - "patch_item_items__item_id__patch", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Patch Item Patch", + "operationId": "patch_item_items__item_id__patch", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], "requestBody": { "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Item" - } + "schema": {"$ref": "#/components/schemas/Item"} } }, "required": True, @@ -365,19 +285,13 @@ openapi_schema = { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -386,26 +300,22 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Trace Item Trace", - "operationId": - "trace_item_items__item_id__trace", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Trace Item Trace", + "operationId": "trace_item_items__item_id__trace", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], }, }, "/items-not-decorated/{item_id}": { @@ -413,19 +323,13 @@ openapi_schema = { "responses": { "200": { "description": "Successful Response", - "content": { - "application/json": { - "schema": {} - } - }, + "content": {"application/json": {"schema": {}}}, }, "403": { "description": "Forbidden", "content": { "application/json": { - "schema": { - "$ref": "#/components/schemas/Response400" - } + "schema": {"$ref": "#/components/schemas/Response400"} } }, }, @@ -434,26 +338,22 @@ openapi_schema = { "content": { "application/json": { "schema": { - "$ref": - "#/components/schemas/HTTPValidationError" + "$ref": "#/components/schemas/HTTPValidationError" } } }, }, }, - "summary": - "Get Not Decorated Get", - "operationId": - "get_not_decorated_items-not-decorated__item_id__get", - "parameters": [{ - "required": True, - "schema": { - "title": "Item_Id", - "type": "string" - }, - "name": "item_id", - "in": "path", - }], + "summary": "Get Not Decorated Get", + "operationId": "get_not_decorated_items-not-decorated__item_id__get", + "parameters": [ + { + "required": True, + "schema": {"title": "Item_Id", "type": "string"}, + "name": "item_id", + "in": "path", + } + ], } }, }, @@ -464,14 +364,8 @@ openapi_schema = { "required": ["name"], "type": "object", "properties": { - "name": { - "title": "Name", - "type": "string" - }, - "price": { - "title": "Price", - "type": "number" - }, + "name": {"title": "Name", "type": "string"}, + "price": {"title": "Price", "type": "number"}, }, }, "Response400": { @@ -480,18 +374,9 @@ openapi_schema = { "required": ["title", "detail", "error_code"], "type": "object", "properties": { - "title": { - "title": "Title", - "type": "string" - }, - "detail": { - "title": "Detail", - "type": "string" - }, - "error_code": { - "title": "Error_Code", - "type": "integer" - }, + "title": {"title": "Title", "type": "string"}, + "detail": {"title": "Detail", "type": "string"}, + "error_code": {"title": "Error_Code", "type": "integer"}, }, }, "ValidationError": { @@ -502,18 +387,10 @@ openapi_schema = { "loc": { "title": "Location", "type": "array", - "items": { - "type": "string" - }, - }, - "msg": { - "title": "Message", - "type": "string" - }, - "type": { - "title": "Error Type", - "type": "string" + "items": {"type": "string"}, }, + "msg": {"title": "Message", "type": "string"}, + "type": {"title": "Error Type", "type": "string"}, }, }, "HTTPValidationError": { @@ -523,9 +400,7 @@ openapi_schema = { "detail": { "title": "Detail", "type": "array", - "items": { - "$ref": "#/components/schemas/ValidationError" - }, + "items": {"$ref": "#/components/schemas/ValidationError"}, } }, }, @@ -555,13 +430,7 @@ def test_get_api_route_not_decorated(): def test_delete(): response = client.delete("/items/foo", json={"name": "Foo"}) assert response.status_code == 200 - assert response.json() == { - "item_id": "foo", - "item": { - "name": "Foo", - "price": None - } - } + assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}} def test_head(): @@ -579,13 +448,7 @@ def test_options(): def test_patch(): response = client.patch("/items/foo", json={"name": "Foo"}) assert response.status_code == 200 - assert response.json() == { - "item_id": "foo", - "item": { - "name": "Foo", - "price": None - } - } + assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}} def test_trace():