mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Formatting according to guide

pull/97/head
Mohammed 6 years ago
parent
c6d28c8209
commit
84a300ef84
6 changed files with 489 additions and 634 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 324
      fastapi/applications.py
  2. 15
      fastapi/openapi/models.py
  3. 20
      fastapi/openapi/utils.py
  4. 387
      fastapi/routing.py
  5. 4
      fastapi/utils.py
  6. 373
      tests/test_additional_responses.py

324
fastapi/applications.py
View File

@ -2,8 +2,8 @@ from typing import Any, Callable, Dict, List, Optional, Type
from fastapi import routing from fastapi import routing
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html 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.models import AdditionalResponse, AdditionalResponseDescription
from fastapi.openapi.utils import get_openapi
from pydantic import BaseModel from pydantic import BaseModel
from starlette.applications import Starlette from starlette.applications import Starlette
from starlette.exceptions import ExceptionMiddleware, HTTPException from starlette.exceptions import ExceptionMiddleware, HTTPException
@ -105,23 +105,23 @@ class FastAPI(Starlette):
self.add_exception_handler(HTTPException, http_exception) self.add_exception_handler(HTTPException, http_exception)
def add_api_route( def add_api_route(
self, self,
path: str, path: str,
endpoint: Callable, endpoint: Callable,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
methods: List[str] = None, methods: List[str] = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> None: ) -> None:
self.router.add_api_route( self.router.add_api_route(
path, path,
@ -142,22 +142,22 @@ class FastAPI(Starlette):
) )
def api_route( def api_route(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
methods: List[str] = None, methods: List[str] = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
def decorator(func: Callable) -> Callable: def decorator(func: Callable) -> Callable:
self.router.add_api_route( self.router.add_api_route(
@ -182,31 +182,33 @@ class FastAPI(Starlette):
return decorator return decorator
def include_router( def include_router(
self, self,
router: routing.APIRouter, router: routing.APIRouter,
*, *,
prefix: str = "", prefix: str = "",
tags: List[str] = None, tags: List[str] = None,
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
) -> None: ) -> 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( def get(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.get( return self.router.get(
path, path,
@ -225,21 +227,21 @@ class FastAPI(Starlette):
) )
def put( def put(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.put( return self.router.put(
path, path,
@ -258,21 +260,21 @@ class FastAPI(Starlette):
) )
def post( def post(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.post( return self.router.post(
path, path,
@ -291,21 +293,21 @@ class FastAPI(Starlette):
) )
def delete( def delete(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.delete( return self.router.delete(
path, path,
@ -324,21 +326,21 @@ class FastAPI(Starlette):
) )
def options( def options(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.options( return self.router.options(
path, path,
@ -357,21 +359,21 @@ class FastAPI(Starlette):
) )
def head( def head(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.head( return self.router.head(
path, path,
@ -390,21 +392,21 @@ class FastAPI(Starlette):
) )
def patch( def patch(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.patch( return self.router.patch(
path, path,
@ -423,21 +425,21 @@ class FastAPI(Starlette):
) )
def trace( def trace(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.router.trace( return self.router.trace(
path, path,

15
fastapi/openapi/models.py
View File

@ -1,10 +1,10 @@
import logging import logging
from enum import Enum 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 import BaseModel, Schema as PSchema
from pydantic.types import UrlStr
from pydantic.fields import Field from pydantic.fields import Field
from pydantic.types import UrlStr
try: try:
import email_validator import email_validator
@ -351,19 +351,12 @@ class BaseAdditionalResponse(BaseModel):
class AdditionalResponse(BaseAdditionalResponse): class AdditionalResponse(BaseAdditionalResponse):
status_code: int = PSchema( 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 # NOTE: waiting for pydantic to allow `typing.Type[BasicModel]` type
# so, going for `Any` and extra validation on # so, going for `Any` and extra validation on
# routing methods # routing methods
models: Optional[List[Any]] = PSchema( models: Optional[List[Any]] = PSchema([], title="Additional Response Models")
[],
title='Additional Response Models',
)
class AdditionalResponseDescription(BaseAdditionalResponse): class AdditionalResponseDescription(BaseAdditionalResponse):

20
fastapi/openapi/utils.py
View File

@ -207,24 +207,22 @@ def get_openapi_path(
} }
for add_response_code, add_response in route.additional_responses.items(): for add_response_code, add_response in route.additional_responses.items():
add_response_schema = {} add_response_schema = {}
if (add_response.content_type or route.content_type.media_type if (
) == 'application/json' and add_response.schema_field is not None: 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_schema(
add_response.schema_field, add_response.schema_field,
model_name_map=model_name_map, model_name_map=model_name_map,
ref_prefix=REF_PREFIX, ref_prefix=REF_PREFIX,
) )
add_content = { add_content = {
add_response.content_type or add_response.content_type
route.content_type.media_type: { or route.content_type.media_type: {"schema": add_response_schema}
"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 path[method.lower()] = operation
return path, security_schemes, definitions return path, security_schemes, definitions

387
fastapi/routing.py
View File

@ -1,14 +1,14 @@
import asyncio import asyncio
import inspect import inspect
import logging 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 import params
from fastapi.dependencies.models import Dependant from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies
from fastapi.encoders import jsonable_encoder from fastapi.encoders import jsonable_encoder
from fastapi.utils import UnconstrainedConfig
from fastapi.openapi.models import AdditionalResponse, AdditionalResponseDescription from fastapi.openapi.models import AdditionalResponse, AdditionalResponseDescription
from fastapi.utils import UnconstrainedConfig
from pydantic import BaseModel, Schema from pydantic import BaseModel, Schema
from pydantic.error_wrappers import ErrorWrapper, ValidationError from pydantic.error_wrappers import ErrorWrapper, ValidationError
from pydantic.fields import Field from pydantic.fields import Field
@ -146,31 +146,30 @@ class APIRoute(routing.Route):
for add_response in additional_responses: for add_response in additional_responses:
if isinstance(add_response, int): if isinstance(add_response, int):
continue 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) existed_codes.append(add_response.status_code)
response_models = [ response_models = [m for m in add_response.models]
m for m in\
add_response.models
]
valid_response_models = True valid_response_models = True
try: try:
valid_response_models = all([ valid_response_models = all(
issubclass(m, BaseModel) [issubclass(m, BaseModel) for m in response_models]
for m in response_models )
])
except TypeError as te: except TypeError as te:
valid_response_models = False valid_response_models = False
if not valid_response_models: if not valid_response_models:
raise ValueError( raise ValueError(
"All response models must be " "All response models must be "
"a subclass of `pydantic.BaseModel` " "a subclass of `pydantic.BaseModel` "
"model.", "model."
) )
if (add_response.content_type == 'application/json' or lenient_issubclass( if add_response.content_type == "application/json" or lenient_issubclass(
content_type, JSONResponse)): content_type, JSONResponse
):
if len(response_models): if len(response_models):
schema_field = Field( schema_field = Field(
name=f'Additional_response_{add_response.status_code}', name=f"Additional_response_{add_response.status_code}",
type_=Union[tuple(response_models)], type_=Union[tuple(response_models)],
class_validators=[], class_validators=[],
default=None, default=None,
@ -186,9 +185,8 @@ class APIRoute(routing.Route):
description=add_response.description, description=add_response.description,
content_type=add_response.content_type, content_type=add_response.content_type,
schema_field=schema_field, schema_field=schema_field,
) )
self.additional_responses[add_response.status_code] = \ self.additional_responses[add_response.status_code] = add_resp_description
add_resp_description
self.deprecated = deprecated self.deprecated = deprecated
if methods is None: if methods is None:
methods = ["GET"] methods = ["GET"]
@ -216,23 +214,23 @@ class APIRoute(routing.Route):
class APIRouter(routing.Router): class APIRouter(routing.Router):
def add_api_route( def add_api_route(
self, self,
path: str, path: str,
endpoint: Callable, endpoint: Callable,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
methods: List[str] = None, methods: List[str] = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> None: ) -> None:
route = APIRoute( route = APIRoute(
path, path,
@ -254,22 +252,22 @@ class APIRouter(routing.Router):
self.routes.append(route) self.routes.append(route)
def api_route( def api_route(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
methods: List[str] = None, methods: List[str] = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
def decorator(func: Callable) -> Callable: def decorator(func: Callable) -> Callable:
self.add_api_route( self.add_api_route(
@ -294,12 +292,12 @@ class APIRouter(routing.Router):
return decorator return decorator
def include_router( def include_router(
self, self,
router: "APIRouter", router: "APIRouter",
*, *,
prefix: str = "", prefix: str = "",
tags: List[str] = None, tags: List[str] = None,
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
) -> None: ) -> None:
if prefix: if prefix:
assert prefix.startswith("/"), "A path prefix must start with '/'" assert prefix.startswith("/"), "A path prefix must start with '/'"
@ -310,20 +308,20 @@ class APIRouter(routing.Router):
if isinstance(route, APIRoute): if isinstance(route, APIRoute):
# really ugly hack and repitition # really ugly hack and repitition
prev_add_resp = route.additional_responses prev_add_resp = route.additional_responses
existed_codes = [422, route.status_code existed_codes = [422, route.status_code] + [
] + [int(c) for c in prev_add_resp.keys()] int(c) for c in prev_add_resp.keys()
]
for add_response in additional_responses: 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) existed_codes.append(add_response.status_code)
response_models = [ response_models = [m for m in add_response.models]
m for m in\
add_response.models
]
valid_response_models = True valid_response_models = True
try: try:
valid_response_models = all([ valid_response_models = all(
issubclass(m, BaseModel) for m in response_models [issubclass(m, BaseModel) for m in response_models]
]) )
except AttributeError as ae: except AttributeError as ae:
valid_response_models = False valid_response_models = False
if not valid_response_models: if not valid_response_models:
@ -332,11 +330,13 @@ class APIRouter(routing.Router):
"a subclass of `pydantic.BaseModel`" "a subclass of `pydantic.BaseModel`"
"model." "model."
) )
if (add_response.content_type == 'application/json' or lenient_issubclass( if (
route.content_type, JSONResponse)): add_response.content_type == "application/json"
or lenient_issubclass(route.content_type, JSONResponse)
):
if len(response_models): if len(response_models):
schema_field = Field( schema_field = Field(
name=f'Additional_response_{add_response.status_code}', name=f"Additional_response_{add_response.status_code}",
type_=Union[tuple(response_models)], type_=Union[tuple(response_models)],
class_validators=[], class_validators=[],
default=None, default=None,
@ -352,9 +352,10 @@ class APIRouter(routing.Router):
description=add_response.description, description=add_response.description,
content_type=add_response.content_type, content_type=add_response.content_type,
schema_field=schema_field, schema_field=schema_field,
) )
route.additional_responses[add_response.status_code] = \ route.additional_responses[
add_resp_description add_response.status_code
] = add_resp_description
self.add_api_route( self.add_api_route(
prefix + route.path, prefix + route.path,
route.endpoint, route.endpoint,
@ -382,21 +383,21 @@ class APIRouter(routing.Router):
) )
def get( def get(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -416,21 +417,21 @@ class APIRouter(routing.Router):
) )
def put( def put(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -450,21 +451,21 @@ class APIRouter(routing.Router):
) )
def post( def post(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -484,21 +485,21 @@ class APIRouter(routing.Router):
) )
def delete( def delete(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -518,21 +519,21 @@ class APIRouter(routing.Router):
) )
def options( def options(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -552,21 +553,21 @@ class APIRouter(routing.Router):
) )
def head( def head(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -586,21 +587,21 @@ class APIRouter(routing.Router):
) )
def patch( def patch(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,
@ -620,21 +621,21 @@ class APIRouter(routing.Router):
) )
def trace( def trace(
self, self,
path: str, path: str,
*, *,
response_model: Type[BaseModel] = None, response_model: Type[BaseModel] = None,
status_code: int = 200, status_code: int = 200,
tags: List[str] = None, tags: List[str] = None,
summary: str = None, summary: str = None,
description: str = None, description: str = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
additional_responses: List[AdditionalResponse] = [], additional_responses: List[AdditionalResponse] = [],
deprecated: bool = None, deprecated: bool = None,
operation_id: str = None, operation_id: str = None,
include_in_schema: bool = True, include_in_schema: bool = True,
content_type: Type[Response] = JSONResponse, content_type: Type[Response] = JSONResponse,
name: str = None, name: str = None,
) -> Callable: ) -> Callable:
return self.api_route( return self.api_route(
path=path, path=path,

4
fastapi/utils.py
View File

@ -33,9 +33,7 @@ def get_flat_models_from_routes(
if route.additional_responses: if route.additional_responses:
for _, add_response in route.additional_responses.items(): for _, add_response in route.additional_responses.items():
if add_response.schema_field is not None: if add_response.schema_field is not None:
responses_from_routes.append( responses_from_routes.append(add_response.schema_field)
add_response.schema_field,
)
flat_models = get_flat_models_from_fields( flat_models = get_flat_models_from_fields(
body_fields_from_routes + responses_from_routes body_fields_from_routes + responses_from_routes
) )

373
tests/test_additional_responses.py
View File

@ -13,28 +13,22 @@ class Item(BaseModel):
class Response400(BaseModel): class Response400(BaseModel):
'''HTTP 4xx Response Schema''' """HTTP 4xx Response Schema"""
title: str title: str
detail: str detail: str
error_code: int # functional error ref error_code: int # functional error ref
response_403 = AdditionalResponse( response_403 = AdditionalResponse(
status_code = 403, status_code=403, description="Forbidden", models=[Response400]
description = 'Forbidden',
models = [
Response400,
],
) )
additional_responses = [ additional_responses = [response_403]
response_403,
]
@app.api_route( @app.api_route(
"/items/{item_id}", "/items/{item_id}", methods=["GET"], additional_responses=additional_responses
methods=["GET"],
additional_responses=additional_responses,
) )
def get_items(item_id: str): def get_items(item_id: str):
return {"item_id": item_id} return {"item_id": item_id}
@ -51,42 +45,27 @@ app.add_api_route(
) )
@app.delete( @app.delete("/items/{item_id}", additional_responses=additional_responses)
"/items/{item_id}",
additional_responses=additional_responses,
)
def delete_item(item_id: str, item: Item): def delete_item(item_id: str, item: Item):
return {"item_id": item_id, "item": item} return {"item_id": item_id, "item": item}
@app.head( @app.head("/items/{item_id}", additional_responses=additional_responses)
"/items/{item_id}",
additional_responses=additional_responses,
)
def head_item(item_id: str): def head_item(item_id: str):
return JSONResponse(headers={"x-fastapi-item-id": item_id}) return JSONResponse(headers={"x-fastapi-item-id": item_id})
@app.options( @app.options("/items/{item_id}", additional_responses=additional_responses)
"/items/{item_id}",
additional_responses=additional_responses,
)
def options_item(item_id: str): def options_item(item_id: str):
return JSONResponse(headers={"x-fastapi-item-id": item_id}) return JSONResponse(headers={"x-fastapi-item-id": item_id})
@app.patch( @app.patch("/items/{item_id}", additional_responses=additional_responses)
"/items/{item_id}",
additional_responses=additional_responses,
)
def patch_item(item_id: str, item: Item): def patch_item(item_id: str, item: Item):
return {"item_id": item_id, "item": item} return {"item_id": item_id, "item": item}
@app.trace( @app.trace("/items/{item_id}", additional_responses=additional_responses)
"/items/{item_id}",
additional_responses=additional_responses,
)
def trace_item(item_id: str): def trace_item(item_id: str):
return JSONResponse(media_type="message/http") return JSONResponse(media_type="message/http")
@ -95,29 +74,20 @@ client = TestClient(app)
openapi_schema = { openapi_schema = {
"openapi": "3.0.2", "openapi": "3.0.2",
"info": { "info": {"title": "Fast API", "version": "0.1.0"},
"title": "Fast API",
"version": "0.1.0"
},
"paths": { "paths": {
"/items/{item_id}": { "/items/{item_id}": {
"get": { "get": {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -126,44 +96,34 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Get Items Get",
"Get Items Get", "operationId": "get_items_items__item_id__get",
"operationId": "parameters": [
"get_items_items__item_id__get", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
}, },
"delete": { "delete": {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -172,32 +132,26 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Delete Item Delete",
"Delete Item Delete", "operationId": "delete_item_items__item_id__delete",
"operationId": "parameters": [
"delete_item_items__item_id__delete", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
"requestBody": { "requestBody": {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Item"}
"$ref": "#/components/schemas/Item"
}
} }
}, },
"required": True, "required": True,
@ -207,19 +161,13 @@ openapi_schema = {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -227,9 +175,7 @@ openapi_schema = {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -238,44 +184,34 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Options Item Options",
"Options Item Options", "operationId": "options_item_items__item_id__options",
"operationId": "parameters": [
"options_item_items__item_id__options", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
}, },
"head": { "head": {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -284,44 +220,34 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Head Item Head",
"Head Item Head", "operationId": "head_item_items__item_id__head",
"operationId": "parameters": [
"head_item_items__item_id__head", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
}, },
"patch": { "patch": {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -330,32 +256,26 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Patch Item Patch",
"Patch Item Patch", "operationId": "patch_item_items__item_id__patch",
"operationId": "parameters": [
"patch_item_items__item_id__patch", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
"requestBody": { "requestBody": {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Item"}
"$ref": "#/components/schemas/Item"
}
} }
}, },
"required": True, "required": True,
@ -365,19 +285,13 @@ openapi_schema = {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -386,26 +300,22 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Trace Item Trace",
"Trace Item Trace", "operationId": "trace_item_items__item_id__trace",
"operationId": "parameters": [
"trace_item_items__item_id__trace", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
}, },
}, },
"/items-not-decorated/{item_id}": { "/items-not-decorated/{item_id}": {
@ -413,19 +323,13 @@ openapi_schema = {
"responses": { "responses": {
"200": { "200": {
"description": "Successful Response", "description": "Successful Response",
"content": { "content": {"application/json": {"schema": {}}},
"application/json": {
"schema": {}
}
},
}, },
"403": { "403": {
"description": "Forbidden", "description": "Forbidden",
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {"$ref": "#/components/schemas/Response400"}
"$ref": "#/components/schemas/Response400"
}
} }
}, },
}, },
@ -434,26 +338,22 @@ openapi_schema = {
"content": { "content": {
"application/json": { "application/json": {
"schema": { "schema": {
"$ref": "$ref": "#/components/schemas/HTTPValidationError"
"#/components/schemas/HTTPValidationError"
} }
} }
}, },
}, },
}, },
"summary": "summary": "Get Not Decorated Get",
"Get Not Decorated Get", "operationId": "get_not_decorated_items-not-decorated__item_id__get",
"operationId": "parameters": [
"get_not_decorated_items-not-decorated__item_id__get", {
"parameters": [{ "required": True,
"required": True, "schema": {"title": "Item_Id", "type": "string"},
"schema": { "name": "item_id",
"title": "Item_Id", "in": "path",
"type": "string" }
}, ],
"name": "item_id",
"in": "path",
}],
} }
}, },
}, },
@ -464,14 +364,8 @@ openapi_schema = {
"required": ["name"], "required": ["name"],
"type": "object", "type": "object",
"properties": { "properties": {
"name": { "name": {"title": "Name", "type": "string"},
"title": "Name", "price": {"title": "Price", "type": "number"},
"type": "string"
},
"price": {
"title": "Price",
"type": "number"
},
}, },
}, },
"Response400": { "Response400": {
@ -480,18 +374,9 @@ openapi_schema = {
"required": ["title", "detail", "error_code"], "required": ["title", "detail", "error_code"],
"type": "object", "type": "object",
"properties": { "properties": {
"title": { "title": {"title": "Title", "type": "string"},
"title": "Title", "detail": {"title": "Detail", "type": "string"},
"type": "string" "error_code": {"title": "Error_Code", "type": "integer"},
},
"detail": {
"title": "Detail",
"type": "string"
},
"error_code": {
"title": "Error_Code",
"type": "integer"
},
}, },
}, },
"ValidationError": { "ValidationError": {
@ -502,18 +387,10 @@ openapi_schema = {
"loc": { "loc": {
"title": "Location", "title": "Location",
"type": "array", "type": "array",
"items": { "items": {"type": "string"},
"type": "string"
},
},
"msg": {
"title": "Message",
"type": "string"
},
"type": {
"title": "Error Type",
"type": "string"
}, },
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
}, },
}, },
"HTTPValidationError": { "HTTPValidationError": {
@ -523,9 +400,7 @@ openapi_schema = {
"detail": { "detail": {
"title": "Detail", "title": "Detail",
"type": "array", "type": "array",
"items": { "items": {"$ref": "#/components/schemas/ValidationError"},
"$ref": "#/components/schemas/ValidationError"
},
} }
}, },
}, },
@ -555,13 +430,7 @@ def test_get_api_route_not_decorated():
def test_delete(): def test_delete():
response = client.delete("/items/foo", json={"name": "Foo"}) response = client.delete("/items/foo", json={"name": "Foo"})
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == { assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}}
"item_id": "foo",
"item": {
"name": "Foo",
"price": None
}
}
def test_head(): def test_head():
@ -579,13 +448,7 @@ def test_options():
def test_patch(): def test_patch():
response = client.patch("/items/foo", json={"name": "Foo"}) response = client.patch("/items/foo", json={"name": "Foo"})
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == { assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}}
"item_id": "foo",
"item": {
"name": "Foo",
"price": None
}
}
def test_trace(): def test_trace():

Write Preview
Loading…
Cancel
Save