Compare commits

..

10 Commits

Author SHA1 Message Date
Sebastián Ramírez
78272ac1f3 🔖 Release 0.39.0 2019-09-29 17:17:44 -05:00
Sebastián Ramírez
f1bee9a271 📝 Update release notes 2019-09-29 17:09:37 -05:00
jonathanunderwood
b20b2218cd Allow defaults in path parameters (and don't use them) (#450) (#464)
This allows using parameters that can have defaults (e.g. `None`) that can be used as query parameters.

But can also be used in routers with that include those parameters as part of the path.
2019-09-29 17:03:16 -05:00
Sebastián Ramírez
b9cf69cd42 📝 Update release notes 2019-09-29 16:50:00 -05:00
toppk
f803c77515 Add support for specifying a default_response_class (#467) 2019-09-29 16:47:35 -05:00
Sebastián Ramírez
0c67022048 📝 Update release notes 2019-09-29 16:24:52 -05:00
dmontagu
d8fe307d61 Add support for strings and __future__ type annotations (#451)
* Add support for strings and __future__ annotations

* Add comments indicating reason for string annotations

* Fix ignores (including removing some unused ignores)
2019-09-29 16:19:09 -05:00
Sebastián Ramírez
580cf8f4e2 🔖 Release 0.38.1 2019-09-01 07:56:37 -05:00
Sebastián Ramírez
af390af77c 📝 Update release notes 2019-09-01 07:53:20 -05:00
Kamal Gill
4642f63a1e 🐛 Use proper import for Request -- fixes #492 (#493) 2019-09-01 07:51:42 -05:00
15 changed files with 668 additions and 56 deletions

View File

@@ -29,6 +29,7 @@ starlette = "==0.12.8"
pydantic = "==0.32.2"
databases = {extras = ["sqlite"],version = "*"}
hypercorn = "*"
orjson = "*"
[requires]
python_version = "3.6"

View File

@@ -1,5 +1,17 @@
## Latest changes
## 0.39.0
* Allow path parameters to have default values (e.g. `None`) and discard them instead of raising an error.
* This allows declaring a parameter like `user_id: str = None` that can be taken from a query parameter, but the same path operation can be included in a router with a path `/users/{user_id}`, in which case will be taken from the path and will be required.
* PR [#464](https://github.com/tiangolo/fastapi/pull/464) by [@jonathanunderwood](https://github.com/jonathanunderwood).
* Add support for setting a `default_response_class` in the `FastAPI` instance or in `include_router`. Initial PR [#467](https://github.com/tiangolo/fastapi/pull/467) by [@toppk](https://github.com/toppk).
* Add support for type annotations using strings and `from __future__ import annotations`. PR [#451](https://github.com/tiangolo/fastapi/pull/451) by [@dmontagu](https://github.com/dmontagu).
## 0.38.1
* Fix incorrect `Request` class import. PR [#493](https://github.com/tiangolo/fastapi/pull/493) by [@kamalgill](https://github.com/kamalgill).
## 0.38.0
* Add recent articles to [External Links](https://fastapi.tiangolo.com/external-links/) and recent opinions. PR [#490](https://github.com/tiangolo/fastapi/pull/490).

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.38.0"
__version__ = "0.39.0"
from starlette.background import BackgroundTasks

View File

@@ -33,11 +33,13 @@ class FastAPI(Starlette):
version: str = "0.1.0",
openapi_url: Optional[str] = "/openapi.json",
openapi_prefix: str = "",
default_response_class: Type[Response] = JSONResponse,
docs_url: Optional[str] = "/docs",
redoc_url: Optional[str] = "/redoc",
swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
**extra: Dict[str, Any],
) -> None:
self.default_response_class = default_response_class
self._debug = debug
self.router: routing.APIRouter = routing.APIRouter(
routes, dependency_overrides_provider=self
@@ -144,7 +146,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> None:
self.router.add_api_route(
@@ -166,7 +168,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -190,7 +192,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
def decorator(func: Callable) -> Callable:
@@ -213,7 +215,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
return func
@@ -240,6 +242,7 @@ class FastAPI(Starlette):
tags: List[str] = None,
dependencies: Sequence[Depends] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
default_response_class: Optional[Type[Response]] = None,
) -> None:
self.router.include_router(
router,
@@ -247,6 +250,8 @@ class FastAPI(Starlette):
tags=tags,
dependencies=dependencies,
responses=responses or {},
default_response_class=default_response_class
or self.default_response_class,
)
def get(
@@ -268,7 +273,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.get(
@@ -288,7 +293,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -311,7 +316,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.put(
@@ -331,7 +336,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -354,7 +359,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.post(
@@ -374,7 +379,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -397,7 +402,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.delete(
@@ -417,7 +422,7 @@ class FastAPI(Starlette):
operation_id=operation_id,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -440,7 +445,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.options(
@@ -460,7 +465,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -483,7 +488,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.head(
@@ -503,7 +508,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -526,7 +531,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.patch(
@@ -546,7 +551,7 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)
@@ -569,7 +574,7 @@ class FastAPI(Starlette):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.router.trace(
@@ -589,6 +594,6 @@ class FastAPI(Starlette):
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
response_class=response_class or self.default_response_class,
name=name,
)

View File

@@ -26,7 +26,7 @@ from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
from pydantic.fields import Field, Required, Shape
from pydantic.schema import get_annotation_from_schema
from pydantic.utils import lenient_issubclass
from pydantic.utils import ForwardRef, evaluate_forwardref, lenient_issubclass
from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
from starlette.datastructures import FormData, Headers, QueryParams, UploadFile
@@ -171,6 +171,30 @@ def is_scalar_sequence_field(field: Field) -> bool:
return False
def get_typed_signature(call: Callable) -> inspect.Signature:
signature = inspect.signature(call)
globalns = getattr(call, "__globals__", {})
typed_params = [
inspect.Parameter(
name=param.name,
kind=param.kind,
default=param.default,
annotation=get_typed_annotation(param, globalns),
)
for param in signature.parameters.values()
]
typed_signature = inspect.Signature(typed_params)
return typed_signature
def get_typed_annotation(param: inspect.Parameter, globalns: Dict[str, Any]) -> Any:
annotation = param.annotation
if isinstance(annotation, str):
annotation = ForwardRef(annotation)
annotation = evaluate_forwardref(annotation, globalns, globalns)
return annotation
def get_dependant(
*,
path: str,
@@ -180,7 +204,7 @@ def get_dependant(
use_cache: bool = True,
) -> Dependant:
path_param_names = get_path_param_names(path)
endpoint_signature = inspect.signature(call)
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
for param_name, param in signature_params.items():
@@ -196,16 +220,18 @@ def get_dependant(
continue
param_field = get_param_field(param=param, default_schema=params.Query)
if param_name in path_param_names:
assert param.default == param.empty or isinstance(
param.default, params.Path
), "Path params must have no defaults or use Path(...)"
assert is_scalar_field(
field=param_field
), f"Path params must be of one of the supported types"
if isinstance(param.default, params.Path):
ignore_default = False
else:
ignore_default = True
param_field = get_param_field(
param=param,
default_schema=params.Path,
force_type=params.ParamTypes.path,
ignore_default=ignore_default,
)
add_param_to_fields(field=param_field, dependant=dependant)
elif is_scalar_field(field=param_field):
@@ -248,10 +274,11 @@ def get_param_field(
param: inspect.Parameter,
default_schema: Type[params.Param] = params.Param,
force_type: params.ParamTypes = None,
ignore_default: bool = False,
) -> Field:
default_value = Required
had_schema = False
if not param.default == param.empty:
if not param.default == param.empty and ignore_default is False:
default_value = param.default
if isinstance(default_value, Schema):
had_schema = True
@@ -329,8 +356,12 @@ async def solve_dependencies(
]:
values: Dict[str, Any] = {}
errors: List[ErrorWrapper] = []
response = response or Response( # type: ignore
content=None, status_code=None, headers=None, media_type=None, background=None
response = response or Response(
content=None,
status_code=None, # type: ignore
headers=None,
media_type=None,
background=None,
)
dependency_cache = dependency_cache or {}
sub_dependant: Dependant
@@ -405,7 +436,7 @@ async def solve_dependencies(
values.update(cookie_values)
errors += path_errors + query_errors + header_errors + cookie_errors
if dependant.body_params:
body_values, body_errors = await request_body_to_args( # type: ignore # body_params checked above
body_values, body_errors = await request_body_to_args( # body_params checked above
required_params=dependant.body_params, received_body=body
)
values.update(body_values)

View File

@@ -2,8 +2,8 @@ from typing import Any, Sequence
from pydantic import ValidationError
from pydantic.error_wrappers import ErrorList
from requests import Request
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.requests import Request
from starlette.websockets import WebSocket

View File

@@ -11,7 +11,7 @@ try:
import email_validator
assert email_validator # make autoflake ignore the unused import
from pydantic.types import EmailStr # type: ignore
from pydantic.types import EmailStr
except ImportError: # pragma: no cover
logger.warning(
"email-validator not installed, email fields will be treated as str.\n"

View File

@@ -151,6 +151,10 @@ def get_openapi_path(
security_schemes: Dict[str, Any] = {}
definitions: Dict[str, Any] = {}
assert route.methods is not None, "Methods must be a list"
assert (
route.response_class and route.response_class.media_type
), "A response class with media_type is needed to generate OpenAPI"
route_response_media_type: str = route.response_class.media_type
if route.include_in_schema:
for method in route.methods:
operation = get_openapi_operation_metadata(route=route, method=method)
@@ -185,7 +189,7 @@ def get_openapi_path(
field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
)
response.setdefault("content", {}).setdefault(
route.response_class.media_type, {}
route_response_media_type, {}
)["schema"] = response_schema
status_text: Optional[str] = status_code_ranges.get(
str(additional_status_code).upper()
@@ -213,7 +217,7 @@ def get_openapi_path(
] = route.response_description
operation.setdefault("responses", {}).setdefault(
status_code, {}
).setdefault("content", {}).setdefault(route.response_class.media_type, {})[
).setdefault("content", {}).setdefault(route_response_media_type, {})[
"schema"
] = response_schema

View File

@@ -200,7 +200,7 @@ class APIRoute(routing.Route):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Optional[Type[Response]] = None,
dependency_overrides_provider: Any = None,
) -> None:
self.path = path
@@ -215,9 +215,6 @@ class APIRoute(routing.Route):
)
self.response_model = response_model
if self.response_model:
assert lenient_issubclass(
response_class, JSONResponse
), "To declare a type the response must be a JSON response"
response_name = "Response_" + self.unique_id
self.response_field: Optional[Field] = Field(
name=response_name,
@@ -299,7 +296,7 @@ class APIRoute(routing.Route):
dependant=self.dependant,
body_field=self.body_field,
status_code=self.status_code,
response_class=self.response_class,
response_class=self.response_class or JSONResponse,
response_field=self.secure_cloned_response_field,
response_model_include=self.response_model_include,
response_model_exclude=self.response_model_exclude,
@@ -346,7 +343,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> None:
route = self.route_class(
@@ -394,7 +391,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
def decorator(func: Callable) -> Callable:
@@ -445,6 +442,7 @@ class APIRouter(routing.Router):
tags: List[str] = None,
dependencies: Sequence[params.Depends] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
default_response_class: Optional[Type[Response]] = None,
) -> None:
if prefix:
assert prefix.startswith("/"), "A path prefix must start with '/'"
@@ -484,7 +482,7 @@ class APIRouter(routing.Router):
response_model_by_alias=route.response_model_by_alias,
response_model_skip_defaults=route.response_model_skip_defaults,
include_in_schema=route.include_in_schema,
response_class=route.response_class,
response_class=route.response_class or default_response_class,
name=route.name,
)
elif isinstance(route, routing.Route):
@@ -523,10 +521,9 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
@@ -568,7 +565,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -612,7 +609,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -656,7 +653,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -700,7 +697,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -744,7 +741,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -788,7 +785,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(
@@ -832,7 +829,7 @@ class APIRouter(routing.Router):
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
response_class: Type[Response] = None,
name: str = None,
) -> Callable:
return self.api_route(

View File

@@ -58,10 +58,10 @@ def create_cloned_field(field: Field) -> Field:
use_type = original_type
if lenient_issubclass(original_type, BaseModel):
original_type = cast(Type[BaseModel], original_type)
use_type = create_model( # type: ignore
use_type = create_model(
original_type.__name__,
__config__=original_type.__config__,
__validators__=original_type.__validators__,
__validators__=original_type.__validators__, # type: ignore
)
for f in original_type.__fields__.values():
use_type.__fields__[f.name] = f

View File

@@ -39,6 +39,7 @@ test = [
"email_validator",
"sqlalchemy",
"databases[sqlite]",
"orjson"
]
doc = [
"mkdocs",

View File

@@ -0,0 +1,216 @@
from typing import Any
import orjson
from fastapi import APIRouter, FastAPI
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse
from starlette.testclient import TestClient
class ORJSONResponse(JSONResponse):
media_type = "application/x-orjson"
def render(self, content: Any) -> bytes:
return orjson.dumps(content)
class OverrideResponse(JSONResponse):
media_type = "application/x-override"
app = FastAPI(default_response_class=ORJSONResponse)
router_a = APIRouter()
router_a_a = APIRouter()
router_a_b_override = APIRouter() # Overrides default class
router_b_override = APIRouter() # Overrides default class
router_b_a = APIRouter()
router_b_a_c_override = APIRouter() # Overrides default class again
@app.get("/")
def get_root():
return {"msg": "Hello World"}
@app.get("/override", response_class=PlainTextResponse)
def get_path_override():
return "Hello World"
@router_a.get("/")
def get_a():
return {"msg": "Hello A"}
@router_a.get("/override", response_class=PlainTextResponse)
def get_a_path_override():
return "Hello A"
@router_a_a.get("/")
def get_a_a():
return {"msg": "Hello A A"}
@router_a_a.get("/override", response_class=PlainTextResponse)
def get_a_a_path_override():
return "Hello A A"
@router_a_b_override.get("/")
def get_a_b():
return "Hello A B"
@router_a_b_override.get("/override", response_class=HTMLResponse)
def get_a_b_path_override():
return "Hello A B"
@router_b_override.get("/")
def get_b():
return "Hello B"
@router_b_override.get("/override", response_class=HTMLResponse)
def get_b_path_override():
return "Hello B"
@router_b_a.get("/")
def get_b_a():
return "Hello B A"
@router_b_a.get("/override", response_class=HTMLResponse)
def get_b_a_path_override():
return "Hello B A"
@router_b_a_c_override.get("/")
def get_b_a_c():
return "Hello B A C"
@router_b_a_c_override.get("/override", response_class=OverrideResponse)
def get_b_a_c_path_override():
return {"msg": "Hello B A C"}
router_b_a.include_router(
router_b_a_c_override, prefix="/c", default_response_class=HTMLResponse
)
router_b_override.include_router(router_b_a, prefix="/a")
router_a.include_router(router_a_a, prefix="/a")
router_a.include_router(
router_a_b_override, prefix="/b", default_response_class=PlainTextResponse
)
app.include_router(router_a, prefix="/a")
app.include_router(
router_b_override, prefix="/b", default_response_class=PlainTextResponse
)
client = TestClient(app)
orjson_type = "application/x-orjson"
text_type = "text/plain; charset=utf-8"
html_type = "text/html; charset=utf-8"
override_type = "application/x-override"
def test_app():
with client:
response = client.get("/")
assert response.json() == {"msg": "Hello World"}
assert response.headers["content-type"] == orjson_type
def test_app_override():
with client:
response = client.get("/override")
assert response.content == b"Hello World"
assert response.headers["content-type"] == text_type
def test_router_a():
with client:
response = client.get("/a")
assert response.json() == {"msg": "Hello A"}
assert response.headers["content-type"] == orjson_type
def test_router_a_override():
with client:
response = client.get("/a/override")
assert response.content == b"Hello A"
assert response.headers["content-type"] == text_type
def test_router_a_a():
with client:
response = client.get("/a/a")
assert response.json() == {"msg": "Hello A A"}
assert response.headers["content-type"] == orjson_type
def test_router_a_a_override():
with client:
response = client.get("/a/a/override")
assert response.content == b"Hello A A"
assert response.headers["content-type"] == text_type
def test_router_a_b():
with client:
response = client.get("/a/b")
assert response.content == b"Hello A B"
assert response.headers["content-type"] == text_type
def test_router_a_b_override():
with client:
response = client.get("/a/b/override")
assert response.content == b"Hello A B"
assert response.headers["content-type"] == html_type
def test_router_b():
with client:
response = client.get("/b")
assert response.content == b"Hello B"
assert response.headers["content-type"] == text_type
def test_router_b_override():
with client:
response = client.get("/b/override")
assert response.content == b"Hello B"
assert response.headers["content-type"] == html_type
def test_router_b_a():
with client:
response = client.get("/b/a")
assert response.content == b"Hello B A"
assert response.headers["content-type"] == text_type
def test_router_b_a_override():
with client:
response = client.get("/b/a/override")
assert response.content == b"Hello B A"
assert response.headers["content-type"] == html_type
def test_router_b_a_c():
with client:
response = client.get("/b/a/c")
assert response.content == b"Hello B A C"
assert response.headers["content-type"] == html_type
def test_router_b_a_c_override():
with client:
response = client.get("/b/a/c/override")
assert response.json() == {"msg": "Hello B A C"}
assert response.headers["content-type"] == override_type

View File

@@ -0,0 +1,206 @@
from fastapi import APIRouter, FastAPI
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse
from starlette.testclient import TestClient
class OverrideResponse(JSONResponse):
media_type = "application/x-override"
app = FastAPI()
router_a = APIRouter()
router_a_a = APIRouter()
router_a_b_override = APIRouter() # Overrides default class
router_b_override = APIRouter() # Overrides default class
router_b_a = APIRouter()
router_b_a_c_override = APIRouter() # Overrides default class again
@app.get("/")
def get_root():
return {"msg": "Hello World"}
@app.get("/override", response_class=PlainTextResponse)
def get_path_override():
return "Hello World"
@router_a.get("/")
def get_a():
return {"msg": "Hello A"}
@router_a.get("/override", response_class=PlainTextResponse)
def get_a_path_override():
return "Hello A"
@router_a_a.get("/")
def get_a_a():
return {"msg": "Hello A A"}
@router_a_a.get("/override", response_class=PlainTextResponse)
def get_a_a_path_override():
return "Hello A A"
@router_a_b_override.get("/")
def get_a_b():
return "Hello A B"
@router_a_b_override.get("/override", response_class=HTMLResponse)
def get_a_b_path_override():
return "Hello A B"
@router_b_override.get("/")
def get_b():
return "Hello B"
@router_b_override.get("/override", response_class=HTMLResponse)
def get_b_path_override():
return "Hello B"
@router_b_a.get("/")
def get_b_a():
return "Hello B A"
@router_b_a.get("/override", response_class=HTMLResponse)
def get_b_a_path_override():
return "Hello B A"
@router_b_a_c_override.get("/")
def get_b_a_c():
return "Hello B A C"
@router_b_a_c_override.get("/override", response_class=OverrideResponse)
def get_b_a_c_path_override():
return {"msg": "Hello B A C"}
router_b_a.include_router(
router_b_a_c_override, prefix="/c", default_response_class=HTMLResponse
)
router_b_override.include_router(router_b_a, prefix="/a")
router_a.include_router(router_a_a, prefix="/a")
router_a.include_router(
router_a_b_override, prefix="/b", default_response_class=PlainTextResponse
)
app.include_router(router_a, prefix="/a")
app.include_router(
router_b_override, prefix="/b", default_response_class=PlainTextResponse
)
client = TestClient(app)
json_type = "application/json"
text_type = "text/plain; charset=utf-8"
html_type = "text/html; charset=utf-8"
override_type = "application/x-override"
def test_app():
with client:
response = client.get("/")
assert response.json() == {"msg": "Hello World"}
assert response.headers["content-type"] == json_type
def test_app_override():
with client:
response = client.get("/override")
assert response.content == b"Hello World"
assert response.headers["content-type"] == text_type
def test_router_a():
with client:
response = client.get("/a")
assert response.json() == {"msg": "Hello A"}
assert response.headers["content-type"] == json_type
def test_router_a_override():
with client:
response = client.get("/a/override")
assert response.content == b"Hello A"
assert response.headers["content-type"] == text_type
def test_router_a_a():
with client:
response = client.get("/a/a")
assert response.json() == {"msg": "Hello A A"}
assert response.headers["content-type"] == json_type
def test_router_a_a_override():
with client:
response = client.get("/a/a/override")
assert response.content == b"Hello A A"
assert response.headers["content-type"] == text_type
def test_router_a_b():
with client:
response = client.get("/a/b")
assert response.content == b"Hello A B"
assert response.headers["content-type"] == text_type
def test_router_a_b_override():
with client:
response = client.get("/a/b/override")
assert response.content == b"Hello A B"
assert response.headers["content-type"] == html_type
def test_router_b():
with client:
response = client.get("/b")
assert response.content == b"Hello B"
assert response.headers["content-type"] == text_type
def test_router_b_override():
with client:
response = client.get("/b/override")
assert response.content == b"Hello B"
assert response.headers["content-type"] == html_type
def test_router_b_a():
with client:
response = client.get("/b/a")
assert response.content == b"Hello B A"
assert response.headers["content-type"] == text_type
def test_router_b_a_override():
with client:
response = client.get("/b/a/override")
assert response.content == b"Hello B A"
assert response.headers["content-type"] == html_type
def test_router_b_a_c():
with client:
response = client.get("/b/a/c")
assert response.content == b"Hello B A C"
assert response.headers["content-type"] == html_type
def test_router_b_a_c_override():
with client:
response = client.get("/b/a/c/override")
assert response.json() == {"msg": "Hello B A C"}
assert response.headers["content-type"] == override_type

View File

@@ -0,0 +1,136 @@
from fastapi import APIRouter, FastAPI
from starlette.testclient import TestClient
app = FastAPI()
user_router = APIRouter()
item_router = APIRouter()
@user_router.get("/")
def get_users():
return [{"user_id": "u1"}, {"user_id": "u2"}]
@user_router.get("/{user_id}")
def get_user(user_id: str):
return {"user_id": user_id}
@item_router.get("/")
def get_items(user_id: str = None):
if user_id is None:
return [{"item_id": "i1", "user_id": "u1"}, {"item_id": "i2", "user_id": "u2"}]
else:
return [{"item_id": "i2", "user_id": user_id}]
@item_router.get("/{item_id}")
def get_item(item_id: str, user_id: str = None):
if user_id is None:
return {"item_id": item_id}
else:
return {"item_id": item_id, "user_id": user_id}
app.include_router(user_router, prefix="/users")
app.include_router(item_router, prefix="/items")
app.include_router(item_router, prefix="/users/{user_id}/items")
client = TestClient(app)
def test_get_users():
"""Check that /users returns expected data"""
response = client.get("/users")
assert response.status_code == 200
assert response.json() == [{"user_id": "u1"}, {"user_id": "u2"}]
def test_get_user():
"""Check that /users/{user_id} returns expected data"""
response = client.get("/users/abc123")
assert response.status_code == 200
assert response.json() == {"user_id": "abc123"}
def test_get_items_1():
"""Check that /items returns expected data"""
response = client.get("/items")
assert response.status_code == 200
assert response.json() == [
{"item_id": "i1", "user_id": "u1"},
{"item_id": "i2", "user_id": "u2"},
]
def test_get_items_2():
"""Check that /items returns expected data with user_id specified"""
response = client.get("/items?user_id=abc123")
assert response.status_code == 200
assert response.json() == [{"item_id": "i2", "user_id": "abc123"}]
def test_get_item_1():
"""Check that /items/{item_id} returns expected data"""
response = client.get("/items/item01")
assert response.status_code == 200
assert response.json() == {"item_id": "item01"}
def test_get_item_2():
"""Check that /items/{item_id} returns expected data with user_id specified"""
response = client.get("/items/item01?user_id=abc123")
assert response.status_code == 200
assert response.json() == {"item_id": "item01", "user_id": "abc123"}
def test_get_users_items():
"""Check that /users/{user_id}/items returns expected data"""
response = client.get("/users/abc123/items")
assert response.status_code == 200
assert response.json() == [{"item_id": "i2", "user_id": "abc123"}]
def test_get_users_item():
"""Check that /users/{user_id}/items returns expected data"""
response = client.get("/users/abc123/items/item01")
assert response.status_code == 200
assert response.json() == {"item_id": "item01", "user_id": "abc123"}
def test_schema_1():
"""Check that the user_id is a required path parameter under /users"""
response = client.get("/openapi.json")
assert response.status_code == 200
r = response.json()
d = {
"required": True,
"schema": {"title": "User_Id", "type": "string"},
"name": "user_id",
"in": "path",
}
assert d in r["paths"]["/users/{user_id}"]["get"]["parameters"]
assert d in r["paths"]["/users/{user_id}/items/"]["get"]["parameters"]
def test_schema_2():
"""Check that the user_id is an optional query parameter under /items"""
response = client.get("/openapi.json")
assert response.status_code == 200
r = response.json()
d = {
"required": False,
"schema": {"title": "User_Id", "type": "string"},
"name": "user_id",
"in": "query",
}
assert d in r["paths"]["/items/{item_id}"]["get"]["parameters"]
assert d in r["paths"]["/items/"]["get"]["parameters"]

View File

@@ -21,18 +21,21 @@ class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(reusable_oauth2)):
# Here we use string annotations to test them
def get_current_user(oauth_header: "str" = Security(reusable_oauth2)):
user = User(username=oauth_header)
return user
@app.post("/login")
def read_current_user(form_data: OAuth2PasswordRequestFormStrict = Depends()):
# Here we use string annotations to test them
def read_current_user(form_data: "OAuth2PasswordRequestFormStrict" = Depends()):
return form_data
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
# Here we use string annotations to test them
def read_current_user(current_user: "User" = Depends(get_current_user)):
return current_user