diff --git a/fastapi/routing.py b/fastapi/routing.py index 9ca2f4673..8cfc576a5 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -17,7 +17,7 @@ from typing import ( Any, Callable, Optional, - Union, + Union, Dict, List, Set, Type, ) from annotated_doc import Doc @@ -484,47 +484,50 @@ class APIWebSocketRoute(routing.WebSocketRoute): class APIRoute(routing.Route): def __init__( - self, - path: str, - endpoint: Callable[..., Any], - *, - response_model: Any = Default(None), - status_code: Optional[int] = None, - tags: Optional[list[Union[str, Enum]]] = None, - dependencies: Optional[Sequence[params.Depends]] = None, - summary: Optional[str] = None, - description: Optional[str] = None, - response_description: str = "Successful Response", - responses: Optional[dict[Union[int, str], dict[str, Any]]] = None, - deprecated: Optional[bool] = None, - name: Optional[str] = None, - methods: Optional[Union[set[str], list[str]]] = None, - operation_id: Optional[str] = None, - response_model_include: Optional[IncEx] = None, - response_model_exclude: Optional[IncEx] = None, - response_model_by_alias: bool = True, - response_model_exclude_unset: bool = False, - response_model_exclude_defaults: bool = False, - response_model_exclude_none: bool = False, - include_in_schema: bool = True, - response_class: Union[type[Response], DefaultPlaceholder] = Default( - JSONResponse - ), - dependency_overrides_provider: Optional[Any] = None, - callbacks: Optional[list[BaseRoute]] = None, - openapi_extra: Optional[dict[str, Any]] = None, - generate_unique_id_function: Union[ - Callable[["APIRoute"], str], DefaultPlaceholder - ] = Default(generate_unique_id), + self, + path: str, + endpoint: Callable[..., Any], + *, + response_model: Any = Default(None), + status_code: Optional[int] = None, + tags: Optional[List[Union[str, Enum]]] = None, + dependencies: Optional[Sequence[params.Depends]] = None, + summary: Optional[str] = None, + description: Optional[str] = None, + response_description: str = "Successful Response", + responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None, + deprecated: Optional[bool] = None, + name: Optional[str] = None, + methods: Optional[Union[Set[str], List[str]]] = None, + operation_id: Optional[str] = None, + response_model_include: Optional[IncEx] = None, + response_model_exclude: Optional[IncEx] = None, + response_model_by_alias: bool = True, + response_model_exclude_unset: bool = False, + response_model_exclude_defaults: bool = False, + response_model_exclude_none: bool = False, + include_in_schema: bool = True, + response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse), + dependency_overrides_provider: Optional[Any] = None, + callbacks: Optional[List[BaseRoute]] = None, + openapi_extra: Optional[Dict[str, Any]] = None, + generate_unique_id_function: Union[Callable[["APIRoute"], str], DefaultPlaceholder] = Default( + generate_unique_id + ), ) -> None: self.path = path 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): return_annotation = get_typed_return_annotation(endpoint) if lenient_issubclass(return_annotation, Response): response_model = None else: response_model = return_annotation + self.response_model = response_model self.summary = summary self.response_description = response_description @@ -546,30 +549,41 @@ class APIRoute(routing.Route): self.responses = responses or {} self.name = get_name(endpoint) if name is None else name self.path_regex, self.path_format, self.param_convertors = compile_path(path) + + # ---- METHODS (with sexy if/else + auto HEAD) ---- if methods is None: 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): - current_generate_unique_id: Callable[[APIRoute], str] = ( - generate_unique_id_function.value - ) + current_generate_unique_id: Callable[["APIRoute"], str] = generate_unique_id_function.value else: current_generate_unique_id = generate_unique_id_function + self.unique_id = self.operation_id or current_generate_unique_id(self) + # normalize enums e.g. http.HTTPStatus if isinstance(status_code, IntEnum): status_code = int(status_code) self.status_code = status_code + if self.response_model: assert is_body_allowed_for_status_code(status_code), ( f"Status code {status_code} must not have a response body" ) 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( name=response_name, 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 # will always be created. # TODO: remove when deprecating Pydantic v1 - self.secure_cloned_response_field: Optional[ModelField] = ( - create_cloned_field(self.response_field) + self.secure_cloned_response_field: Optional[ModelField] = create_cloned_field( + self.response_field ) else: self.response_field = None # type: ignore self.secure_cloned_response_field = None + self.dependencies = list(dependencies or []) self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "") + # if a "form feed" character (page break) is found in the description text, # truncate description text to the content preceding the first "form feed" self.description = self.description.split("\f")[0].strip() + response_fields = {} for additional_status_code, response in self.responses.items(): 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" ) 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( - name=response_name, type_=model, mode="serialization" + name=response_name, + type_=model, + mode="serialization", ) response_fields[additional_status_code] = response_field + if response_fields: - self.response_fields: dict[Union[int, str], ModelField] = response_fields + self.response_fields: Dict[Union[int, str], ModelField] = response_fields else: self.response_fields = {} - assert callable(endpoint), "An endpoint must be a callable" 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]: self.dependant.dependencies.insert( 0, get_parameterless_sub_dependant(depends=depends, path=self.path_format), ) + self._flat_dependant = get_flat_dependant(self.dependant) - self._embed_body_fields = _should_embed_body_fields( - self._flat_dependant.body_params - ) + self._embed_body_fields = _should_embed_body_fields(self._flat_dependant.body_params) self.body_field = get_body_field( flat_dependant=self._flat_dependant, name=self.unique_id, embed_body_fields=self._embed_body_fields, ) + self.app = request_response(self.get_route_handler()) def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]: