pydantic = "==0.32.2"
databases = {extras = ["sqlite"],version = "*"}
hypercorn = "*"
+orjson = "*"
[requires]
python_version = "3.6"
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
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(
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,
)
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:
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
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,
tags=tags,
dependencies=dependencies,
responses=responses or {},
+ default_response_class=default_response_class
+ or self.default_response_class,
)
def get(
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(
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,
)
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(
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,
)
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(
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,
)
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(
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,
)
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(
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,
)
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(
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,
)
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(
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,
)
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(
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,
)
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)
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()
] = 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
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
)
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,
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,
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(
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:
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 '/'"
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):
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,
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(
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(
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(
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(
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(
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(
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(
"email_validator",
"sqlalchemy",
"databases[sqlite]",
+ "orjson"
]
doc = [
"mkdocs",
--- /dev/null
+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
--- /dev/null
+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
projects / thirdparty / fastapi / fastapi.git / commitdiff
