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

demo: logic change

pull/14770/head
Maksymilian Majer 6 months ago
parent
8c32e91c10
commit
f73ef2410d
1 changed files with 73 additions and 56 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. 129
      fastapi/routing.py

129
fastapi/routing.py
View File

@ -17,7 +17,7 @@ from typing import (
Any, Any,
Callable, Callable,
Optional, Optional,
Union, Union, Dict, List, Set, Type,
) )
from annotated_doc import Doc from annotated_doc import Doc
@ -484,47 +484,50 @@ class APIWebSocketRoute(routing.WebSocketRoute):
class APIRoute(routing.Route): class APIRoute(routing.Route):
def __init__( def __init__(
self, self,
path: str, path: str,
endpoint: Callable[..., Any], endpoint: Callable[..., Any],
*, *,
response_model: Any = Default(None), response_model: Any = Default(None),
status_code: Optional[int] = None, status_code: Optional[int] = None,
tags: Optional[list[Union[str, Enum]]] = None, tags: Optional[List[Union[str, Enum]]] = None,
dependencies: Optional[Sequence[params.Depends]] = None, dependencies: Optional[Sequence[params.Depends]] = None,
summary: Optional[str] = None, summary: Optional[str] = None,
description: Optional[str] = None, description: Optional[str] = None,
response_description: str = "Successful Response", response_description: str = "Successful Response",
responses: Optional[dict[Union[int, str], dict[str, Any]]] = None, responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
deprecated: Optional[bool] = None, deprecated: Optional[bool] = None,
name: Optional[str] = None, name: Optional[str] = None,
methods: Optional[Union[set[str], list[str]]] = None, methods: Optional[Union[Set[str], List[str]]] = None,
operation_id: Optional[str] = None, operation_id: Optional[str] = None,
response_model_include: Optional[IncEx] = None, response_model_include: Optional[IncEx] = None,
response_model_exclude: Optional[IncEx] = None, response_model_exclude: Optional[IncEx] = None,
response_model_by_alias: bool = True, response_model_by_alias: bool = True,
response_model_exclude_unset: bool = False, response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False, response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False, response_model_exclude_none: bool = False,
include_in_schema: bool = True, include_in_schema: bool = True,
response_class: Union[type[Response], DefaultPlaceholder] = Default( response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse),
JSONResponse dependency_overrides_provider: Optional[Any] = None,
), callbacks: Optional[List[BaseRoute]] = None,
dependency_overrides_provider: Optional[Any] = None, openapi_extra: Optional[Dict[str, Any]] = None,
callbacks: Optional[list[BaseRoute]] = None, generate_unique_id_function: Union[Callable[["APIRoute"], str], DefaultPlaceholder] = Default(
openapi_extra: Optional[dict[str, Any]] = None, generate_unique_id
generate_unique_id_function: Union[ ),
Callable[["APIRoute"], str], DefaultPlaceholder
] = Default(generate_unique_id),
) -> None: ) -> None:
self.path = path self.path = path
self.endpoint = endpoint self.endpoint = endpoint
# NEW: fail fast before we inspect annotations / build response fields
assert callable(endpoint), "An endpoint must be a callable"
if isinstance(response_model, DefaultPlaceholder): if isinstance(response_model, DefaultPlaceholder):
return_annotation = get_typed_return_annotation(endpoint) return_annotation = get_typed_return_annotation(endpoint)
if lenient_issubclass(return_annotation, Response): if lenient_issubclass(return_annotation, Response):
response_model = None response_model = None
else: else:
response_model = return_annotation response_model = return_annotation
self.response_model = response_model self.response_model = response_model
self.summary = summary self.summary = summary
self.response_description = response_description self.response_description = response_description
@ -546,30 +549,41 @@ class APIRoute(routing.Route):
self.responses = responses or {} self.responses = responses or {}
self.name = get_name(endpoint) if name is None else name self.name = get_name(endpoint) if name is None else name
self.path_regex, self.path_format, self.param_convertors = compile_path(path) self.path_regex, self.path_format, self.param_convertors = compile_path(path)
# ---- METHODS (with sexy if/else + auto HEAD) ----
if methods is None: if methods is None:
methods = ["GET"] methods = ["GET"]
self.methods: set[str] = {method.upper() for method in methods} else:
# Ensure it's mutable so we can append
methods = list(methods)
normalized_methods = {m.upper() for m in methods}
# NEW: add HEAD if GET exists (unless already provided)
if "GET" in normalized_methods and "HEAD" not in normalized_methods:
methods.append("HEAD")
normalized_methods.add("HEAD")
self.methods: Set[str] = normalized_methods
# -----------------------------------------------
if isinstance(generate_unique_id_function, DefaultPlaceholder): if isinstance(generate_unique_id_function, DefaultPlaceholder):
current_generate_unique_id: Callable[[APIRoute], str] = ( current_generate_unique_id: Callable[["APIRoute"], str] = generate_unique_id_function.value
generate_unique_id_function.value
)
else: else:
current_generate_unique_id = generate_unique_id_function current_generate_unique_id = generate_unique_id_function
self.unique_id = self.operation_id or current_generate_unique_id(self) self.unique_id = self.operation_id or current_generate_unique_id(self)
# normalize enums e.g. http.HTTPStatus # normalize enums e.g. http.HTTPStatus
if isinstance(status_code, IntEnum): if isinstance(status_code, IntEnum):
status_code = int(status_code) status_code = int(status_code)
self.status_code = status_code self.status_code = status_code
if self.response_model: if self.response_model:
assert is_body_allowed_for_status_code(status_code), ( assert is_body_allowed_for_status_code(status_code), (
f"Status code {status_code} must not have a response body" f"Status code {status_code} must not have a response body"
) )
response_name = "Response_" + self.unique_id response_name = "Response_" + self.unique_id
if annotation_is_pydantic_v1(self.response_model):
raise PydanticV1NotSupportedError(
"pydantic.v1 models are no longer supported by FastAPI."
f" Please update the response model {self.response_model!r}."
)
self.response_field = create_model_field( self.response_field = create_model_field(
name=response_name, name=response_name,
type_=self.response_model, type_=self.response_model,
@ -583,17 +597,20 @@ class APIRoute(routing.Route):
# By being a new field, no inheritance will be passed as is. A new model # By being a new field, no inheritance will be passed as is. A new model
# will always be created. # will always be created.
# TODO: remove when deprecating Pydantic v1 # TODO: remove when deprecating Pydantic v1
self.secure_cloned_response_field: Optional[ModelField] = ( self.secure_cloned_response_field: Optional[ModelField] = create_cloned_field(
create_cloned_field(self.response_field) self.response_field
) )
else: else:
self.response_field = None # type: ignore self.response_field = None # type: ignore
self.secure_cloned_response_field = None self.secure_cloned_response_field = None
self.dependencies = list(dependencies or []) self.dependencies = list(dependencies or [])
self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "") self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "")
# if a "form feed" character (page break) is found in the description text, # if a "form feed" character (page break) is found in the description text,
# truncate description text to the content preceding the first "form feed" # truncate description text to the content preceding the first "form feed"
self.description = self.description.split("\f")[0].strip() self.description = self.description.split("\f")[0].strip()
response_fields = {} response_fields = {}
for additional_status_code, response in self.responses.items(): for additional_status_code, response in self.responses.items():
assert isinstance(response, dict), "An additional response must be a dict" assert isinstance(response, dict), "An additional response must be a dict"
@ -603,38 +620,38 @@ class APIRoute(routing.Route):
f"Status code {additional_status_code} must not have a response body" f"Status code {additional_status_code} must not have a response body"
) )
response_name = f"Response_{additional_status_code}_{self.unique_id}" response_name = f"Response_{additional_status_code}_{self.unique_id}"
if annotation_is_pydantic_v1(model):
raise PydanticV1NotSupportedError(
"pydantic.v1 models are no longer supported by FastAPI."
f" In responses={{}}, please update {model}."
)
response_field = create_model_field( response_field = create_model_field(
name=response_name, type_=model, mode="serialization" name=response_name,
type_=model,
mode="serialization",
) )
response_fields[additional_status_code] = response_field response_fields[additional_status_code] = response_field
if response_fields: if response_fields:
self.response_fields: dict[Union[int, str], ModelField] = response_fields self.response_fields: Dict[Union[int, str], ModelField] = response_fields
else: else:
self.response_fields = {} self.response_fields = {}
assert callable(endpoint), "An endpoint must be a callable"
self.dependant = get_dependant( self.dependant = get_dependant(
path=self.path_format, call=self.endpoint, scope="function" path=self.path_format,
call=self.endpoint,
scope="function",
) )
for depends in self.dependencies[::-1]: for depends in self.dependencies[::-1]:
self.dependant.dependencies.insert( self.dependant.dependencies.insert(
0, 0,
get_parameterless_sub_dependant(depends=depends, path=self.path_format), get_parameterless_sub_dependant(depends=depends, path=self.path_format),
) )
self._flat_dependant = get_flat_dependant(self.dependant) self._flat_dependant = get_flat_dependant(self.dependant)
self._embed_body_fields = _should_embed_body_fields( self._embed_body_fields = _should_embed_body_fields(self._flat_dependant.body_params)
self._flat_dependant.body_params
)
self.body_field = get_body_field( self.body_field = get_body_field(
flat_dependant=self._flat_dependant, flat_dependant=self._flat_dependant,
name=self.unique_id, name=self.unique_id,
embed_body_fields=self._embed_body_fields, embed_body_fields=self._embed_body_fields,
) )
self.app = request_response(self.get_route_handler()) self.app = request_response(self.get_route_handler())
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]: def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:

Write Preview
Loading…
Cancel
Save