-
+
diff --git a/docs_src/request_form_models/tutorial001.py b/docs_src/request_form_models/tutorial001.py
new file mode 100644
index 000000000..98feff0b9
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001.py
@@ -0,0 +1,14 @@
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: FormData = Form()):
+ return data
diff --git a/docs_src/request_form_models/tutorial001_an.py b/docs_src/request_form_models/tutorial001_an.py
new file mode 100644
index 000000000..30483d445
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001_an.py
@@ -0,0 +1,15 @@
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: Annotated[FormData, Form()]):
+ return data
diff --git a/docs_src/request_form_models/tutorial001_an_py39.py b/docs_src/request_form_models/tutorial001_an_py39.py
new file mode 100644
index 000000000..7cc81aae9
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001_an_py39.py
@@ -0,0 +1,16 @@
+from typing import Annotated
+
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: Annotated[FormData, Form()]):
+ return data
diff --git a/fastapi/__init__.py b/fastapi/__init__.py
index ac2508d89..f785f81cd 100644
--- a/fastapi/__init__.py
+++ b/fastapi/__init__.py
@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
-__version__ = "0.112.2"
+__version__ = "0.113.0"
from starlette import status as status
diff --git a/fastapi/_compat.py b/fastapi/_compat.py
index 06b847b4f..f940d6597 100644
--- a/fastapi/_compat.py
+++ b/fastapi/_compat.py
@@ -279,6 +279,12 @@ if PYDANTIC_V2:
BodyModel: Type[BaseModel] = create_model(model_name, **field_params) # type: ignore[call-overload]
return BodyModel
+ def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
+ return [
+ ModelField(field_info=field_info, name=name)
+ for name, field_info in model.model_fields.items()
+ ]
+
else:
from fastapi.openapi.constants import REF_PREFIX as REF_PREFIX
from pydantic import AnyUrl as Url # noqa: F401
@@ -513,6 +519,9 @@ else:
BodyModel.__fields__[f.name] = f # type: ignore[index]
return BodyModel
+ def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
+ return list(model.__fields__.values()) # type: ignore[attr-defined]
+
def _regenerate_error_with_loc(
*, errors: Sequence[Any], loc_prefix: Tuple[Union[str, int], ...]
@@ -532,6 +541,12 @@ def _annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
def field_annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
+ origin = get_origin(annotation)
+ if origin is Union or origin is UnionType:
+ for arg in get_args(annotation):
+ if field_annotation_is_sequence(arg):
+ return True
+ return False
return _annotation_is_sequence(annotation) or _annotation_is_sequence(
get_origin(annotation)
)
diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py
index 0dcba62f1..98ce17b55 100644
--- a/fastapi/dependencies/utils.py
+++ b/fastapi/dependencies/utils.py
@@ -33,6 +33,7 @@ from fastapi._compat import (
field_annotation_is_scalar,
get_annotation_from_field_info,
get_missing_field_error,
+ get_model_fields,
is_bytes_field,
is_bytes_sequence_field,
is_scalar_field,
@@ -56,10 +57,17 @@ from fastapi.security.base import SecurityBase
from fastapi.security.oauth2 import OAuth2, SecurityScopes
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.utils import create_model_field, get_path_param_names
+from pydantic import BaseModel
from pydantic.fields import FieldInfo
from starlette.background import BackgroundTasks as StarletteBackgroundTasks
from starlette.concurrency import run_in_threadpool
-from starlette.datastructures import FormData, Headers, QueryParams, UploadFile
+from starlette.datastructures import (
+ FormData,
+ Headers,
+ ImmutableMultiDict,
+ QueryParams,
+ UploadFile,
+)
from starlette.requests import HTTPConnection, Request
from starlette.responses import Response
from starlette.websockets import WebSocket
@@ -282,7 +290,7 @@ def get_dependant(
), f"Cannot specify multiple FastAPI annotations for {param_name!r}"
continue
assert param_details.field is not None
- if is_body_param(param_field=param_details.field, is_path_param=is_path_param):
+ if isinstance(param_details.field.field_info, params.Body):
dependant.body_params.append(param_details.field)
else:
add_param_to_fields(field=param_details.field, dependant=dependant)
@@ -466,29 +474,16 @@ def analyze_param(
required=field_info.default in (Required, Undefined),
field_info=field_info,
)
+ if is_path_param:
+ assert is_scalar_field(
+ field=field
+ ), "Path params must be of one of the supported types"
+ elif isinstance(field_info, params.Query):
+ assert is_scalar_field(field) or is_scalar_sequence_field(field)
return ParamDetails(type_annotation=type_annotation, depends=depends, field=field)
-def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
- if is_path_param:
- assert is_scalar_field(
- field=param_field
- ), "Path params must be of one of the supported types"
- return False
- elif is_scalar_field(field=param_field):
- return False
- elif isinstance(
- param_field.field_info, (params.Query, params.Header)
- ) and is_scalar_sequence_field(param_field):
- return False
- else:
- assert isinstance(
- param_field.field_info, params.Body
- ), f"Param: {param_field.name} can only be a request body, using Body()"
- return True
-
-
def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None:
field_info = field.field_info
field_info_in = getattr(field_info, "in_", None)
@@ -557,6 +552,7 @@ async def solve_dependencies(
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[Dict[Tuple[Callable[..., Any], Tuple[str]], Any]] = None,
async_exit_stack: AsyncExitStack,
+ embed_body_fields: bool,
) -> SolvedDependency:
values: Dict[str, Any] = {}
errors: List[Any] = []
@@ -598,6 +594,7 @@ async def solve_dependencies(
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
background_tasks = solved_result.background_tasks
dependency_cache.update(solved_result.dependency_cache)
@@ -640,7 +637,9 @@ async def solve_dependencies(
body_values,
body_errors,
) = await request_body_to_args( # body_params checked above
- required_params=dependant.body_params, received_body=body
+ body_fields=dependant.body_params,
+ received_body=body,
+ embed_body_fields=embed_body_fields,
)
values.update(body_values)
errors.extend(body_errors)
@@ -669,138 +668,194 @@ async def solve_dependencies(
)
+def _validate_value_with_model_field(
+ *, field: ModelField, value: Any, values: Dict[str, Any], loc: Tuple[str, ...]
+) -> Tuple[Any, List[Any]]:
+ if value is None:
+ if field.required:
+ return None, [get_missing_field_error(loc=loc)]
+ else:
+ return deepcopy(field.default), []
+ v_, errors_ = field.validate(value, values, loc=loc)
+ if isinstance(errors_, ErrorWrapper):
+ return None, [errors_]
+ elif isinstance(errors_, list):
+ new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
+ return None, new_errors
+ else:
+ return v_, []
+
+
+def _get_multidict_value(field: ModelField, values: Mapping[str, Any]) -> Any:
+ if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)):
+ value = values.getlist(field.alias)
+ else:
+ value = values.get(field.alias, None)
+ if (
+ value is None
+ or (
+ isinstance(field.field_info, params.Form)
+ and isinstance(value, str) # For type checks
+ and value == ""
+ )
+ or (is_sequence_field(field) and len(value) == 0)
+ ):
+ if field.required:
+ return
+ else:
+ return deepcopy(field.default)
+ return value
+
+
def request_params_to_args(
- required_params: Sequence[ModelField],
+ fields: Sequence[ModelField],
received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[Any]]:
- values = {}
+ values: Dict[str, Any] = {}
errors = []
- for field in required_params:
- if is_scalar_sequence_field(field) and isinstance(
- received_params, (QueryParams, Headers)
- ):
- value = received_params.getlist(field.alias) or field.default
- else:
- value = received_params.get(field.alias)
+ for field in fields:
+ value = _get_multidict_value(field, received_params)
field_info = field.field_info
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
loc = (field_info.in_.value, field.alias)
- if value is None:
- if field.required:
- errors.append(get_missing_field_error(loc=loc))
- else:
- values[field.name] = deepcopy(field.default)
- continue
- v_, errors_ = field.validate(value, values, loc=loc)
- if isinstance(errors_, ErrorWrapper):
- errors.append(errors_)
- elif isinstance(errors_, list):
- new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
- errors.extend(new_errors)
+ v_, errors_ = _validate_value_with_model_field(
+ field=field, value=value, values=values, loc=loc
+ )
+ if errors_:
+ errors.extend(errors_)
else:
values[field.name] = v_
return values, errors
+def _should_embed_body_fields(fields: List[ModelField]) -> bool:
+ if not fields:
+ return False
+ # More than one dependency could have the same field, it would show up as multiple
+ # fields but it's the same one, so count them by name
+ body_param_names_set = {field.name for field in fields}
+ # A top level field has to be a single field, not multiple
+ if len(body_param_names_set) > 1:
+ return True
+ first_field = fields[0]
+ # If it explicitly specifies it is embedded, it has to be embedded
+ if getattr(first_field.field_info, "embed", None):
+ return True
+ # If it's a Form (or File) field, it has to be a BaseModel to be top level
+ # otherwise it has to be embedded, so that the key value pair can be extracted
+ if isinstance(first_field.field_info, params.Form) and not lenient_issubclass(
+ first_field.type_, BaseModel
+ ):
+ return True
+ return False
+
+
+async def _extract_form_body(
+ body_fields: List[ModelField],
+ received_body: FormData,
+) -> Dict[str, Any]:
+ values = {}
+ first_field = body_fields[0]
+ first_field_info = first_field.field_info
+
+ for field in body_fields:
+ value = _get_multidict_value(field, received_body)
+ if (
+ isinstance(first_field_info, params.File)
+ and is_bytes_field(field)
+ and isinstance(value, UploadFile)
+ ):
+ value = await value.read()
+ elif (
+ is_bytes_sequence_field(field)
+ and isinstance(first_field_info, params.File)
+ and value_is_sequence(value)
+ ):
+ # For types
+ assert isinstance(value, sequence_types) # type: ignore[arg-type]
+ results: List[Union[bytes, str]] = []
+
+ async def process_fn(
+ fn: Callable[[], Coroutine[Any, Any, Any]],
+ ) -> None:
+ result = await fn()
+ results.append(result) # noqa: B023
+
+ async with anyio.create_task_group() as tg:
+ for sub_value in value:
+ tg.start_soon(process_fn, sub_value.read)
+ value = serialize_sequence_value(field=field, value=results)
+ if value is not None:
+ values[field.name] = value
+ return values
+
+
async def request_body_to_args(
- required_params: List[ModelField],
+ body_fields: List[ModelField],
received_body: Optional[Union[Dict[str, Any], FormData]],
+ embed_body_fields: bool,
) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
- values = {}
+ values: Dict[str, Any] = {}
errors: List[Dict[str, Any]] = []
- if required_params:
- field = required_params[0]
- field_info = field.field_info
- embed = getattr(field_info, "embed", None)
- field_alias_omitted = len(required_params) == 1 and not embed
- if field_alias_omitted:
- received_body = {field.alias: received_body}
-
- for field in required_params:
- loc: Tuple[str, ...]
- if field_alias_omitted:
- loc = ("body",)
- else:
- loc = ("body", field.alias)
-
- value: Optional[Any] = None
- if received_body is not None:
- if (is_sequence_field(field)) and isinstance(received_body, FormData):
- value = received_body.getlist(field.alias)
- else:
- try:
- value = received_body.get(field.alias)
- except AttributeError:
- errors.append(get_missing_field_error(loc))
- continue
- if (
- value is None
- or (isinstance(field_info, params.Form) and value == "")
- or (
- isinstance(field_info, params.Form)
- and is_sequence_field(field)
- and len(value) == 0
- )
- ):
- if field.required:
- errors.append(get_missing_field_error(loc))
- else:
- values[field.name] = deepcopy(field.default)
+ assert body_fields, "request_body_to_args() should be called with fields"
+ single_not_embedded_field = len(body_fields) == 1 and not embed_body_fields
+ first_field = body_fields[0]
+ body_to_process = received_body
+
+ fields_to_extract: List[ModelField] = body_fields
+
+ if single_not_embedded_field and lenient_issubclass(first_field.type_, BaseModel):
+ fields_to_extract = get_model_fields(first_field.type_)
+
+ if isinstance(received_body, FormData):
+ body_to_process = await _extract_form_body(fields_to_extract, received_body)
+
+ if single_not_embedded_field:
+ loc: Tuple[str, ...] = ("body",)
+ v_, errors_ = _validate_value_with_model_field(
+ field=first_field, value=body_to_process, values=values, loc=loc
+ )
+ return {first_field.name: v_}, errors_
+ for field in body_fields:
+ loc = ("body", field.alias)
+ value: Optional[Any] = None
+ if body_to_process is not None:
+ try:
+ value = body_to_process.get(field.alias)
+ # If the received body is a list, not a dict
+ except AttributeError:
+ errors.append(get_missing_field_error(loc))
continue
- if (
- isinstance(field_info, params.File)
- and is_bytes_field(field)
- and isinstance(value, UploadFile)
- ):
- value = await value.read()
- elif (
- is_bytes_sequence_field(field)
- and isinstance(field_info, params.File)
- and value_is_sequence(value)
- ):
- # For types
- assert isinstance(value, sequence_types) # type: ignore[arg-type]
- results: List[Union[bytes, str]] = []
-
- async def process_fn(
- fn: Callable[[], Coroutine[Any, Any, Any]],
- ) -> None:
- result = await fn()
- results.append(result) # noqa: B023
-
- async with anyio.create_task_group() as tg:
- for sub_value in value:
- tg.start_soon(process_fn, sub_value.read)
- value = serialize_sequence_value(field=field, value=results)
-
- v_, errors_ = field.validate(value, values, loc=loc)
-
- if isinstance(errors_, list):
- errors.extend(errors_)
- elif errors_:
- errors.append(errors_)
- else:
- values[field.name] = v_
+ v_, errors_ = _validate_value_with_model_field(
+ field=field, value=value, values=values, loc=loc
+ )
+ if errors_:
+ errors.extend(errors_)
+ else:
+ values[field.name] = v_
return values, errors
-def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]:
- flat_dependant = get_flat_dependant(dependant)
+def get_body_field(
+ *, flat_dependant: Dependant, name: str, embed_body_fields: bool
+) -> Optional[ModelField]:
+ """
+ Get a ModelField representing the request body for a path operation, combining
+ all body parameters into a single field if necessary.
+
+ Used to check if it's form data (with `isinstance(body_field, params.Form)`)
+ or JSON and to generate the JSON Schema for a request body.
+
+ This is **not** used to validate/parse the request body, that's done with each
+ individual body parameter.
+ """
if not flat_dependant.body_params:
return None
first_param = flat_dependant.body_params[0]
- field_info = first_param.field_info
- embed = getattr(field_info, "embed", None)
- body_param_names_set = {param.name for param in flat_dependant.body_params}
- if len(body_param_names_set) == 1 and not embed:
+ if not embed_body_fields:
return first_param
- # If one field requires to embed, all have to be embedded
- # in case a sub-dependency is evaluated with a single unique body field
- # That is combined (embedded) with other body fields
- for param in flat_dependant.body_params:
- setattr(param.field_info, "embed", True) # noqa: B010
model_name = "Body_" + name
BodyModel = create_body_model(
fields=flat_dependant.body_params, model_name=model_name
diff --git a/fastapi/param_functions.py b/fastapi/param_functions.py
index 0d5f27af4..7ddaace25 100644
--- a/fastapi/param_functions.py
+++ b/fastapi/param_functions.py
@@ -1282,7 +1282,7 @@ def Body( # noqa: N802
),
] = _Unset,
embed: Annotated[
- bool,
+ Union[bool, None],
Doc(
"""
When `embed` is `True`, the parameter will be expected in a JSON body as a
@@ -1294,7 +1294,7 @@ def Body( # noqa: N802
[FastAPI docs for Body - Multiple Parameters](https://fastapi.tiangolo.com/tutorial/body-multiple-params/#embed-a-single-body-parameter).
"""
),
- ] = False,
+ ] = None,
media_type: Annotated[
str,
Doc(
diff --git a/fastapi/params.py b/fastapi/params.py
index cc2a5c13c..3dfa5a1a3 100644
--- a/fastapi/params.py
+++ b/fastapi/params.py
@@ -479,7 +479,7 @@ class Body(FieldInfo):
*,
default_factory: Union[Callable[[], Any], None] = _Unset,
annotation: Optional[Any] = None,
- embed: bool = False,
+ embed: Union[bool, None] = None,
media_type: str = "application/json",
alias: Optional[str] = None,
alias_priority: Union[int, None] = _Unset,
@@ -642,7 +642,6 @@ class Form(Body):
default=default,
default_factory=default_factory,
annotation=annotation,
- embed=True,
media_type=media_type,
alias=alias,
alias_priority=alias_priority,
diff --git a/fastapi/routing.py b/fastapi/routing.py
index 2b68cb0d1..7daec9b0f 100644
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@@ -33,8 +33,10 @@ from fastapi._compat import (
from fastapi.datastructures import Default, DefaultPlaceholder
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import (
+ _should_embed_body_fields,
get_body_field,
get_dependant,
+ get_flat_dependant,
get_parameterless_sub_dependant,
get_typed_return_annotation,
solve_dependencies,
@@ -225,6 +227,7 @@ def get_request_handler(
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
dependency_overrides_provider: Optional[Any] = None,
+ embed_body_fields: bool = False,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = asyncio.iscoroutinefunction(dependant.call)
@@ -291,6 +294,7 @@ def get_request_handler(
body=body,
dependency_overrides_provider=dependency_overrides_provider,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
errors = solved_result.errors
if not errors:
@@ -354,7 +358,9 @@ def get_request_handler(
def get_websocket_app(
- dependant: Dependant, dependency_overrides_provider: Optional[Any] = None
+ dependant: Dependant,
+ dependency_overrides_provider: Optional[Any] = None,
+ embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
async def app(websocket: WebSocket) -> None:
async with AsyncExitStack() as async_exit_stack:
@@ -367,6 +373,7 @@ def get_websocket_app(
dependant=dependant,
dependency_overrides_provider=dependency_overrides_provider,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
if solved_result.errors:
raise WebSocketRequestValidationError(
@@ -399,11 +406,15 @@ class APIWebSocketRoute(routing.WebSocketRoute):
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.app = websocket_session(
get_websocket_app(
dependant=self.dependant,
dependency_overrides_provider=dependency_overrides_provider,
+ embed_body_fields=self._embed_body_fields,
)
)
@@ -544,7 +555,15 @@ class APIRoute(routing.Route):
0,
get_parameterless_sub_dependant(depends=depends, path=self.path_format),
)
- self.body_field = get_body_field(dependant=self.dependant, name=self.unique_id)
+ self._flat_dependant = get_flat_dependant(self.dependant)
+ 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]]:
@@ -561,6 +580,7 @@ class APIRoute(routing.Route):
response_model_exclude_defaults=self.response_model_exclude_defaults,
response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider,
+ embed_body_fields=self._embed_body_fields,
)
def matches(self, scope: Scope) -> Tuple[Match, Scope]:
diff --git a/requirements-tests.txt b/requirements-tests.txt
index 08561d23a..de5fdb8a2 100644
--- a/requirements-tests.txt
+++ b/requirements-tests.txt
@@ -3,7 +3,7 @@
pytest >=7.1.3,<8.0.0
coverage[toml] >= 6.5.0,< 8.0
mypy ==1.8.0
-ruff ==0.6.1
+ruff ==0.6.3
dirty-equals ==0.6.0
# TODO: once removing databases from tutorial, upgrade SQLAlchemy
# probably when including SQLModel
diff --git a/scripts/playwright/request_form_models/image01.py b/scripts/playwright/request_form_models/image01.py
new file mode 100644
index 000000000..15bd3858c
--- /dev/null
+++ b/scripts/playwright/request_form_models/image01.py
@@ -0,0 +1,36 @@
+import subprocess
+import time
+
+import httpx
+from playwright.sync_api import Playwright, sync_playwright
+
+
+# Run playwright codegen to generate the code below, copy paste the sections in run()
+def run(playwright: Playwright) -> None:
+ browser = playwright.chromium.launch(headless=False)
+ context = browser.new_context()
+ page = context.new_page()
+ page.goto("http://localhost:8000/docs")
+ page.get_by_role("button", name="POST /login/ Login").click()
+ page.get_by_role("button", name="Try it out").click()
+ page.screenshot(path="docs/en/docs/img/tutorial/request-form-models/image01.png")
+
+ # ---------------------
+ context.close()
+ browser.close()
+
+
+process = subprocess.Popen(
+ ["fastapi", "run", "docs_src/request_form_models/tutorial001.py"]
+)
+try:
+ for _ in range(3):
+ try:
+ response = httpx.get("http://localhost:8000/docs")
+ except httpx.ConnectError:
+ time.sleep(1)
+ break
+ with sync_playwright() as playwright:
+ run(playwright)
+finally:
+ process.terminate()
diff --git a/tests/test_compat.py b/tests/test_compat.py
index bf268b860..270475bf3 100644
--- a/tests/test_compat.py
+++ b/tests/test_compat.py
@@ -1,11 +1,13 @@
-from typing import List, Union
+from typing import Any, Dict, List, Union
from fastapi import FastAPI, UploadFile
from fastapi._compat import (
ModelField,
Undefined,
_get_model_config,
+ get_model_fields,
is_bytes_sequence_annotation,
+ is_scalar_field,
is_uploadfile_sequence_annotation,
)
from fastapi.testclient import TestClient
@@ -91,3 +93,12 @@ def test_is_uploadfile_sequence_annotation():
# and other types, but I'm not even sure it's a good idea to support it as a first
# class "feature"
assert is_uploadfile_sequence_annotation(Union[List[str], List[UploadFile]])
+
+
+def test_is_pv1_scalar_field():
+ # For coverage
+ class Model(BaseModel):
+ foo: Union[str, Dict[str, Any]]
+
+ fields = get_model_fields(Model)
+ assert not is_scalar_field(fields[0])
diff --git a/tests/test_forms_single_model.py b/tests/test_forms_single_model.py
new file mode 100644
index 000000000..7ed3ba3a2
--- /dev/null
+++ b/tests/test_forms_single_model.py
@@ -0,0 +1,129 @@
+from typing import List, Optional
+
+from dirty_equals import IsDict
+from fastapi import FastAPI, Form
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+class FormModel(BaseModel):
+ username: str
+ lastname: str
+ age: Optional[int] = None
+ tags: List[str] = ["foo", "bar"]
+
+
+@app.post("/form/")
+def post_form(user: Annotated[FormModel, Form()]):
+ return user
+
+
+client = TestClient(app)
+
+
+def test_send_all_data():
+ response = client.post(
+ "/form/",
+ data={
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": "70",
+ "tags": ["plumbus", "citadel"],
+ },
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": 70,
+ "tags": ["plumbus", "citadel"],
+ }
+
+
+def test_defaults():
+ response = client.post("/form/", data={"username": "Rick", "lastname": "Sanchez"})
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": None,
+ "tags": ["foo", "bar"],
+ }
+
+
+def test_invalid_data():
+ response = client.post(
+ "/form/",
+ data={
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": "seventy",
+ "tags": ["plumbus", "citadel"],
+ },
+ )
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "int_parsing",
+ "loc": ["body", "age"],
+ "msg": "Input should be a valid integer, unable to parse string as an integer",
+ "input": "seventy",
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "age"],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ }
+ ]
+ }
+ )
+
+
+def test_no_data():
+ response = client.post("/form/")
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"tags": ["foo", "bar"]},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "lastname"],
+ "msg": "Field required",
+ "input": {"tags": ["foo", "bar"]},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "lastname"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
diff --git a/tests/test_forms_single_param.py b/tests/test_forms_single_param.py
new file mode 100644
index 000000000..3bb951441
--- /dev/null
+++ b/tests/test_forms_single_param.py
@@ -0,0 +1,99 @@
+from fastapi import FastAPI, Form
+from fastapi.testclient import TestClient
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+@app.post("/form/")
+def post_form(username: Annotated[str, Form()]):
+ return username
+
+
+client = TestClient(app)
+
+
+def test_single_form_field():
+ response = client.post("/form/", data={"username": "Rick"})
+ assert response.status_code == 200, response.text
+ assert response.json() == "Rick"
+
+
+def test_openapi_schema():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/form/": {
+ "post": {
+ "summary": "Post Form",
+ "operationId": "post_form_form__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {
+ "$ref": "#/components/schemas/Body_post_form_form__post"
+ }
+ }
+ },
+ "required": True,
+ },
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "Body_post_form_form__post": {
+ "properties": {"username": {"type": "string", "title": "Username"}},
+ "type": "object",
+ "required": ["username"],
+ "title": "Body_post_form_form__post",
+ },
+ "HTTPValidationError": {
+ "properties": {
+ "detail": {
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ "type": "array",
+ "title": "Detail",
+ }
+ },
+ "type": "object",
+ "title": "HTTPValidationError",
+ },
+ "ValidationError": {
+ "properties": {
+ "loc": {
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ "type": "array",
+ "title": "Location",
+ },
+ "msg": {"type": "string", "title": "Message"},
+ "type": {"type": "string", "title": "Error Type"},
+ },
+ "type": "object",
+ "required": ["loc", "msg", "type"],
+ "title": "ValidationError",
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/__init__.py b/tests/test_tutorial/test_request_form_models/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001.py b/tests/test_tutorial/test_request_form_models/test_tutorial001.py
new file mode 100644
index 000000000..46c130ee8
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001.py
@@ -0,0 +1,232 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001 import app
+
+ client = TestClient(app)
+ return client
+
+
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py b/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py
new file mode 100644
index 000000000..4e14d89c8
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py
@@ -0,0 +1,232 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001_an import app
+
+ client = TestClient(app)
+ return client
+
+
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py b/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py
new file mode 100644
index 000000000..2e6426aa7
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py
@@ -0,0 +1,240 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+from tests.utils import needs_py39
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001_an_py39 import app
+
+ client = TestClient(app)
+ return client
+
+
+@needs_py39
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+@needs_py39
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }
-
+
diff --git a/docs/en/data/external_links.yml b/docs/en/data/external_links.yml
index 15f6169ee..63fd3d0cf 100644
--- a/docs/en/data/external_links.yml
+++ b/docs/en/data/external_links.yml
@@ -264,6 +264,14 @@ Articles:
author_link: https://devonray.com
link: https://devonray.com/blog/deploying-a-fastapi-project-using-aws-lambda-aurora-cdk
title: Deployment using Docker, Lambda, Aurora, CDK & GH Actions
+ - author: Shubhendra Kushwaha
+ author_link: https://www.linkedin.com/in/theshubhendra/
+ link: https://theshubhendra.medium.com/mastering-soft-delete-advanced-sqlalchemy-techniques-4678f4738947
+ title: 'Mastering Soft Delete: Advanced SQLAlchemy Techniques'
+ - author: Shubhendra Kushwaha
+ author_link: https://www.linkedin.com/in/theshubhendra/
+ link: https://theshubhendra.medium.com/role-based-row-filtering-advanced-sqlalchemy-techniques-733e6b1328f6
+ title: 'Role based row filtering: Advanced SQLAlchemy Techniques'
German:
- author: Marcel Sander (actidoo)
author_link: https://www.actidoo.com
diff --git a/docs/en/data/sponsors.yml b/docs/en/data/sponsors.yml
index 3a767b6b1..d96646fb3 100644
--- a/docs/en/data/sponsors.yml
+++ b/docs/en/data/sponsors.yml
@@ -17,7 +17,7 @@ gold:
- url: https://www.propelauth.com/?utm_source=fastapi&utm_campaign=1223&utm_medium=mainbadge
title: Auth, user management and more for your B2B product
img: https://fastapi.tiangolo.com/img/sponsors/propelauth.png
- - url: https://docs.withcoherence.com/coherence-templates/full-stack-template/#fastapi?utm_medium=advertising&utm_source=fastapi&utm_campaign=docs
+ - url: https://www.withcoherence.com/?utm_medium=advertising&utm_source=fastapi&utm_campaign=website
title: Coherence
img: https://fastapi.tiangolo.com/img/sponsors/coherence.png
- url: https://www.mongodb.com/developer/languages/python/python-quickstart-fastapi/?utm_campaign=fastapi_framework&utm_source=fastapi_sponsorship&utm_medium=web_referral
diff --git a/docs/en/docs/deployment/cloud.md b/docs/en/docs/deployment/cloud.md
index 3ea5087f8..41ada859d 100644
--- a/docs/en/docs/deployment/cloud.md
+++ b/docs/en/docs/deployment/cloud.md
@@ -14,4 +14,4 @@ You might want to try their services and follow their guides:
* Platform.sh
* Porter
-* Coherence
+* Coherence
diff --git a/docs/en/docs/img/tutorial/request-form-models/image01.png b/docs/en/docs/img/tutorial/request-form-models/image01.png
new file mode 100644
index 000000000..3fe32c03d
Binary files /dev/null and b/docs/en/docs/img/tutorial/request-form-models/image01.png differ
diff --git a/docs/en/docs/release-notes.md b/docs/en/docs/release-notes.md
index df4927752..0571523bf 100644
--- a/docs/en/docs/release-notes.md
+++ b/docs/en/docs/release-notes.md
@@ -7,6 +7,58 @@ hide:
## Latest Changes
+## 0.113.0
+
+Now you can declare form fields with Pydantic models:
+
+```python
+from typing import Annotated
+
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: Annotated[FormData, Form()]):
+ return data
+```
+
+Read the new docs: [Form Models](https://fastapi.tiangolo.com/tutorial/request-form-models/).
+
+### Features
+
+* ✨ Add support for Pydantic models in `Form` parameters. PR [#12129](https://github.com/fastapi/fastapi/pull/12129) by [@tiangolo](https://github.com/tiangolo).
+
+### Internal
+
+* 🔧 Update sponsors: Coherence link. PR [#12130](https://github.com/fastapi/fastapi/pull/12130) by [@tiangolo](https://github.com/tiangolo).
+
+## 0.112.4
+
+This release is mainly a big internal refactor to enable adding support for Pydantic models for `Form` fields, but that feature comes in the next release.
+
+This release shouldn't affect apps using FastAPI in any way. You don't even have to upgrade to this version yet. It's just a checkpoint. 🤓
+
+### Refactors
+
+* ♻️ Refactor deciding if `embed` body fields, do not overwrite fields, compute once per router, refactor internals in preparation for Pydantic models in `Form`, `Query` and others. PR [#12117](https://github.com/fastapi/fastapi/pull/12117) by [@tiangolo](https://github.com/tiangolo).
+
+### Internal
+
+* ⏪️ Temporarily revert "✨ Add support for Pydantic models in `Form` parameters" to make a checkpoint release. PR [#12128](https://github.com/fastapi/fastapi/pull/12128) by [@tiangolo](https://github.com/tiangolo). Restored by PR [#12129](https://github.com/fastapi/fastapi/pull/12129).
+* ✨ Add support for Pydantic models in `Form` parameters. PR [#12127](https://github.com/fastapi/fastapi/pull/12127) by [@tiangolo](https://github.com/tiangolo). Reverted by PR [#12128](https://github.com/fastapi/fastapi/pull/12128) to make a checkpoint release with only refactors. Restored by PR [#12129](https://github.com/fastapi/fastapi/pull/12129).
+
+## 0.112.3
+
+This release is mainly internal refactors, it shouldn't affect apps using FastAPI in any way. You don't even have to upgrade to this version yet. There are a few bigger releases coming right after. 🚀
+
### Refactors
* ♻️ Refactor internal `check_file_field()`, rename to `ensure_multipart_is_installed()` to clarify its purpose. PR [#12106](https://github.com/fastapi/fastapi/pull/12106) by [@tiangolo](https://github.com/tiangolo).
@@ -17,6 +69,7 @@ hide:
### Docs
+* 📝 Add External Link: Techniques and applications of SQLAlchemy global filters in FastAPI. PR [#12109](https://github.com/fastapi/fastapi/pull/12109) by [@TheShubhendra](https://github.com/TheShubhendra).
* 📝 Add note about `time.perf_counter()` in middlewares. PR [#12095](https://github.com/fastapi/fastapi/pull/12095) by [@tiangolo](https://github.com/tiangolo).
* 📝 Tweak middleware code sample `time.time()` to `time.perf_counter()`. PR [#11957](https://github.com/fastapi/fastapi/pull/11957) by [@domdent](https://github.com/domdent).
* 🔧 Update sponsors: Coherence. PR [#12093](https://github.com/fastapi/fastapi/pull/12093) by [@tiangolo](https://github.com/tiangolo).
@@ -38,6 +91,7 @@ hide:
### Internal
+* ⬆ [pre-commit.ci] pre-commit autoupdate. PR [#12115](https://github.com/fastapi/fastapi/pull/12115) by [@pre-commit-ci[bot]](https://github.com/apps/pre-commit-ci).
* ⬆ Bump pypa/gh-action-pypi-publish from 1.10.0 to 1.10.1. PR [#12120](https://github.com/fastapi/fastapi/pull/12120) by [@dependabot[bot]](https://github.com/apps/dependabot).
* ⬆ Bump pillow from 10.3.0 to 10.4.0. PR [#12105](https://github.com/fastapi/fastapi/pull/12105) by [@dependabot[bot]](https://github.com/apps/dependabot).
* 💚 Set `include-hidden-files` to `True` when using the `upload-artifact` GH action. PR [#12118](https://github.com/fastapi/fastapi/pull/12118) by [@svlandeg](https://github.com/svlandeg).
diff --git a/docs/en/docs/tutorial/request-form-models.md b/docs/en/docs/tutorial/request-form-models.md
new file mode 100644
index 000000000..8bb1ffb1f
--- /dev/null
+++ b/docs/en/docs/tutorial/request-form-models.md
@@ -0,0 +1,65 @@
+# Form Models
+
+You can use Pydantic models to declare form fields in FastAPI.
+
+/// info
+
+To use forms, first install `python-multipart`.
+
+Make sure you create a [virtual environment](../virtual-environments.md){.internal-link target=_blank}, activate it, and then install it, for example:
+
+```console
+$ pip install python-multipart
+```
+
+///
+
+/// note
+
+This is supported since FastAPI version `0.113.0`. 🤓
+
+///
+
+## Pydantic Models for Forms
+
+You just need to declare a Pydantic model with the fields you want to receive as form fields, and then declare the parameter as `Form`:
+
+//// tab | Python 3.9+
+
+```Python hl_lines="9-11 15"
+{!> ../../../docs_src/request_form_models/tutorial001_an_py39.py!}
+```
+
+////
+
+//// tab | Python 3.8+
+
+```Python hl_lines="8-10 14"
+{!> ../../../docs_src/request_form_models/tutorial001_an.py!}
+```
+
+////
+
+//// tab | Python 3.8+ non-Annotated
+
+/// tip
+
+Prefer to use the `Annotated` version if possible.
+
+///
+
+```Python hl_lines="7-9 13"
+{!> ../../../docs_src/request_form_models/tutorial001.py!}
+```
+
+////
+
+FastAPI will extract the data for each field from the form data in the request and give you the Pydantic model you defined.
+
+## Check the Docs
+
+You can verify it in the docs UI at `/docs`:
+
+
+
diff --git a/docs_src/request_form_models/tutorial001.py b/docs_src/request_form_models/tutorial001.py
new file mode 100644
index 000000000..98feff0b9
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001.py
@@ -0,0 +1,14 @@
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: FormData = Form()):
+ return data
diff --git a/docs_src/request_form_models/tutorial001_an.py b/docs_src/request_form_models/tutorial001_an.py
new file mode 100644
index 000000000..30483d445
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001_an.py
@@ -0,0 +1,15 @@
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: Annotated[FormData, Form()]):
+ return data
diff --git a/docs_src/request_form_models/tutorial001_an_py39.py b/docs_src/request_form_models/tutorial001_an_py39.py
new file mode 100644
index 000000000..7cc81aae9
--- /dev/null
+++ b/docs_src/request_form_models/tutorial001_an_py39.py
@@ -0,0 +1,16 @@
+from typing import Annotated
+
+from fastapi import FastAPI, Form
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class FormData(BaseModel):
+ username: str
+ password: str
+
+
+@app.post("/login/")
+async def login(data: Annotated[FormData, Form()]):
+ return data
diff --git a/fastapi/__init__.py b/fastapi/__init__.py
index ac2508d89..f785f81cd 100644
--- a/fastapi/__init__.py
+++ b/fastapi/__init__.py
@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
-__version__ = "0.112.2"
+__version__ = "0.113.0"
from starlette import status as status
diff --git a/fastapi/_compat.py b/fastapi/_compat.py
index 06b847b4f..f940d6597 100644
--- a/fastapi/_compat.py
+++ b/fastapi/_compat.py
@@ -279,6 +279,12 @@ if PYDANTIC_V2:
BodyModel: Type[BaseModel] = create_model(model_name, **field_params) # type: ignore[call-overload]
return BodyModel
+ def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
+ return [
+ ModelField(field_info=field_info, name=name)
+ for name, field_info in model.model_fields.items()
+ ]
+
else:
from fastapi.openapi.constants import REF_PREFIX as REF_PREFIX
from pydantic import AnyUrl as Url # noqa: F401
@@ -513,6 +519,9 @@ else:
BodyModel.__fields__[f.name] = f # type: ignore[index]
return BodyModel
+ def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
+ return list(model.__fields__.values()) # type: ignore[attr-defined]
+
def _regenerate_error_with_loc(
*, errors: Sequence[Any], loc_prefix: Tuple[Union[str, int], ...]
@@ -532,6 +541,12 @@ def _annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
def field_annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
+ origin = get_origin(annotation)
+ if origin is Union or origin is UnionType:
+ for arg in get_args(annotation):
+ if field_annotation_is_sequence(arg):
+ return True
+ return False
return _annotation_is_sequence(annotation) or _annotation_is_sequence(
get_origin(annotation)
)
diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py
index 0dcba62f1..98ce17b55 100644
--- a/fastapi/dependencies/utils.py
+++ b/fastapi/dependencies/utils.py
@@ -33,6 +33,7 @@ from fastapi._compat import (
field_annotation_is_scalar,
get_annotation_from_field_info,
get_missing_field_error,
+ get_model_fields,
is_bytes_field,
is_bytes_sequence_field,
is_scalar_field,
@@ -56,10 +57,17 @@ from fastapi.security.base import SecurityBase
from fastapi.security.oauth2 import OAuth2, SecurityScopes
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.utils import create_model_field, get_path_param_names
+from pydantic import BaseModel
from pydantic.fields import FieldInfo
from starlette.background import BackgroundTasks as StarletteBackgroundTasks
from starlette.concurrency import run_in_threadpool
-from starlette.datastructures import FormData, Headers, QueryParams, UploadFile
+from starlette.datastructures import (
+ FormData,
+ Headers,
+ ImmutableMultiDict,
+ QueryParams,
+ UploadFile,
+)
from starlette.requests import HTTPConnection, Request
from starlette.responses import Response
from starlette.websockets import WebSocket
@@ -282,7 +290,7 @@ def get_dependant(
), f"Cannot specify multiple FastAPI annotations for {param_name!r}"
continue
assert param_details.field is not None
- if is_body_param(param_field=param_details.field, is_path_param=is_path_param):
+ if isinstance(param_details.field.field_info, params.Body):
dependant.body_params.append(param_details.field)
else:
add_param_to_fields(field=param_details.field, dependant=dependant)
@@ -466,29 +474,16 @@ def analyze_param(
required=field_info.default in (Required, Undefined),
field_info=field_info,
)
+ if is_path_param:
+ assert is_scalar_field(
+ field=field
+ ), "Path params must be of one of the supported types"
+ elif isinstance(field_info, params.Query):
+ assert is_scalar_field(field) or is_scalar_sequence_field(field)
return ParamDetails(type_annotation=type_annotation, depends=depends, field=field)
-def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
- if is_path_param:
- assert is_scalar_field(
- field=param_field
- ), "Path params must be of one of the supported types"
- return False
- elif is_scalar_field(field=param_field):
- return False
- elif isinstance(
- param_field.field_info, (params.Query, params.Header)
- ) and is_scalar_sequence_field(param_field):
- return False
- else:
- assert isinstance(
- param_field.field_info, params.Body
- ), f"Param: {param_field.name} can only be a request body, using Body()"
- return True
-
-
def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None:
field_info = field.field_info
field_info_in = getattr(field_info, "in_", None)
@@ -557,6 +552,7 @@ async def solve_dependencies(
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[Dict[Tuple[Callable[..., Any], Tuple[str]], Any]] = None,
async_exit_stack: AsyncExitStack,
+ embed_body_fields: bool,
) -> SolvedDependency:
values: Dict[str, Any] = {}
errors: List[Any] = []
@@ -598,6 +594,7 @@ async def solve_dependencies(
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
background_tasks = solved_result.background_tasks
dependency_cache.update(solved_result.dependency_cache)
@@ -640,7 +637,9 @@ async def solve_dependencies(
body_values,
body_errors,
) = await request_body_to_args( # body_params checked above
- required_params=dependant.body_params, received_body=body
+ body_fields=dependant.body_params,
+ received_body=body,
+ embed_body_fields=embed_body_fields,
)
values.update(body_values)
errors.extend(body_errors)
@@ -669,138 +668,194 @@ async def solve_dependencies(
)
+def _validate_value_with_model_field(
+ *, field: ModelField, value: Any, values: Dict[str, Any], loc: Tuple[str, ...]
+) -> Tuple[Any, List[Any]]:
+ if value is None:
+ if field.required:
+ return None, [get_missing_field_error(loc=loc)]
+ else:
+ return deepcopy(field.default), []
+ v_, errors_ = field.validate(value, values, loc=loc)
+ if isinstance(errors_, ErrorWrapper):
+ return None, [errors_]
+ elif isinstance(errors_, list):
+ new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
+ return None, new_errors
+ else:
+ return v_, []
+
+
+def _get_multidict_value(field: ModelField, values: Mapping[str, Any]) -> Any:
+ if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)):
+ value = values.getlist(field.alias)
+ else:
+ value = values.get(field.alias, None)
+ if (
+ value is None
+ or (
+ isinstance(field.field_info, params.Form)
+ and isinstance(value, str) # For type checks
+ and value == ""
+ )
+ or (is_sequence_field(field) and len(value) == 0)
+ ):
+ if field.required:
+ return
+ else:
+ return deepcopy(field.default)
+ return value
+
+
def request_params_to_args(
- required_params: Sequence[ModelField],
+ fields: Sequence[ModelField],
received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[Any]]:
- values = {}
+ values: Dict[str, Any] = {}
errors = []
- for field in required_params:
- if is_scalar_sequence_field(field) and isinstance(
- received_params, (QueryParams, Headers)
- ):
- value = received_params.getlist(field.alias) or field.default
- else:
- value = received_params.get(field.alias)
+ for field in fields:
+ value = _get_multidict_value(field, received_params)
field_info = field.field_info
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
loc = (field_info.in_.value, field.alias)
- if value is None:
- if field.required:
- errors.append(get_missing_field_error(loc=loc))
- else:
- values[field.name] = deepcopy(field.default)
- continue
- v_, errors_ = field.validate(value, values, loc=loc)
- if isinstance(errors_, ErrorWrapper):
- errors.append(errors_)
- elif isinstance(errors_, list):
- new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
- errors.extend(new_errors)
+ v_, errors_ = _validate_value_with_model_field(
+ field=field, value=value, values=values, loc=loc
+ )
+ if errors_:
+ errors.extend(errors_)
else:
values[field.name] = v_
return values, errors
+def _should_embed_body_fields(fields: List[ModelField]) -> bool:
+ if not fields:
+ return False
+ # More than one dependency could have the same field, it would show up as multiple
+ # fields but it's the same one, so count them by name
+ body_param_names_set = {field.name for field in fields}
+ # A top level field has to be a single field, not multiple
+ if len(body_param_names_set) > 1:
+ return True
+ first_field = fields[0]
+ # If it explicitly specifies it is embedded, it has to be embedded
+ if getattr(first_field.field_info, "embed", None):
+ return True
+ # If it's a Form (or File) field, it has to be a BaseModel to be top level
+ # otherwise it has to be embedded, so that the key value pair can be extracted
+ if isinstance(first_field.field_info, params.Form) and not lenient_issubclass(
+ first_field.type_, BaseModel
+ ):
+ return True
+ return False
+
+
+async def _extract_form_body(
+ body_fields: List[ModelField],
+ received_body: FormData,
+) -> Dict[str, Any]:
+ values = {}
+ first_field = body_fields[0]
+ first_field_info = first_field.field_info
+
+ for field in body_fields:
+ value = _get_multidict_value(field, received_body)
+ if (
+ isinstance(first_field_info, params.File)
+ and is_bytes_field(field)
+ and isinstance(value, UploadFile)
+ ):
+ value = await value.read()
+ elif (
+ is_bytes_sequence_field(field)
+ and isinstance(first_field_info, params.File)
+ and value_is_sequence(value)
+ ):
+ # For types
+ assert isinstance(value, sequence_types) # type: ignore[arg-type]
+ results: List[Union[bytes, str]] = []
+
+ async def process_fn(
+ fn: Callable[[], Coroutine[Any, Any, Any]],
+ ) -> None:
+ result = await fn()
+ results.append(result) # noqa: B023
+
+ async with anyio.create_task_group() as tg:
+ for sub_value in value:
+ tg.start_soon(process_fn, sub_value.read)
+ value = serialize_sequence_value(field=field, value=results)
+ if value is not None:
+ values[field.name] = value
+ return values
+
+
async def request_body_to_args(
- required_params: List[ModelField],
+ body_fields: List[ModelField],
received_body: Optional[Union[Dict[str, Any], FormData]],
+ embed_body_fields: bool,
) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
- values = {}
+ values: Dict[str, Any] = {}
errors: List[Dict[str, Any]] = []
- if required_params:
- field = required_params[0]
- field_info = field.field_info
- embed = getattr(field_info, "embed", None)
- field_alias_omitted = len(required_params) == 1 and not embed
- if field_alias_omitted:
- received_body = {field.alias: received_body}
-
- for field in required_params:
- loc: Tuple[str, ...]
- if field_alias_omitted:
- loc = ("body",)
- else:
- loc = ("body", field.alias)
-
- value: Optional[Any] = None
- if received_body is not None:
- if (is_sequence_field(field)) and isinstance(received_body, FormData):
- value = received_body.getlist(field.alias)
- else:
- try:
- value = received_body.get(field.alias)
- except AttributeError:
- errors.append(get_missing_field_error(loc))
- continue
- if (
- value is None
- or (isinstance(field_info, params.Form) and value == "")
- or (
- isinstance(field_info, params.Form)
- and is_sequence_field(field)
- and len(value) == 0
- )
- ):
- if field.required:
- errors.append(get_missing_field_error(loc))
- else:
- values[field.name] = deepcopy(field.default)
+ assert body_fields, "request_body_to_args() should be called with fields"
+ single_not_embedded_field = len(body_fields) == 1 and not embed_body_fields
+ first_field = body_fields[0]
+ body_to_process = received_body
+
+ fields_to_extract: List[ModelField] = body_fields
+
+ if single_not_embedded_field and lenient_issubclass(first_field.type_, BaseModel):
+ fields_to_extract = get_model_fields(first_field.type_)
+
+ if isinstance(received_body, FormData):
+ body_to_process = await _extract_form_body(fields_to_extract, received_body)
+
+ if single_not_embedded_field:
+ loc: Tuple[str, ...] = ("body",)
+ v_, errors_ = _validate_value_with_model_field(
+ field=first_field, value=body_to_process, values=values, loc=loc
+ )
+ return {first_field.name: v_}, errors_
+ for field in body_fields:
+ loc = ("body", field.alias)
+ value: Optional[Any] = None
+ if body_to_process is not None:
+ try:
+ value = body_to_process.get(field.alias)
+ # If the received body is a list, not a dict
+ except AttributeError:
+ errors.append(get_missing_field_error(loc))
continue
- if (
- isinstance(field_info, params.File)
- and is_bytes_field(field)
- and isinstance(value, UploadFile)
- ):
- value = await value.read()
- elif (
- is_bytes_sequence_field(field)
- and isinstance(field_info, params.File)
- and value_is_sequence(value)
- ):
- # For types
- assert isinstance(value, sequence_types) # type: ignore[arg-type]
- results: List[Union[bytes, str]] = []
-
- async def process_fn(
- fn: Callable[[], Coroutine[Any, Any, Any]],
- ) -> None:
- result = await fn()
- results.append(result) # noqa: B023
-
- async with anyio.create_task_group() as tg:
- for sub_value in value:
- tg.start_soon(process_fn, sub_value.read)
- value = serialize_sequence_value(field=field, value=results)
-
- v_, errors_ = field.validate(value, values, loc=loc)
-
- if isinstance(errors_, list):
- errors.extend(errors_)
- elif errors_:
- errors.append(errors_)
- else:
- values[field.name] = v_
+ v_, errors_ = _validate_value_with_model_field(
+ field=field, value=value, values=values, loc=loc
+ )
+ if errors_:
+ errors.extend(errors_)
+ else:
+ values[field.name] = v_
return values, errors
-def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]:
- flat_dependant = get_flat_dependant(dependant)
+def get_body_field(
+ *, flat_dependant: Dependant, name: str, embed_body_fields: bool
+) -> Optional[ModelField]:
+ """
+ Get a ModelField representing the request body for a path operation, combining
+ all body parameters into a single field if necessary.
+
+ Used to check if it's form data (with `isinstance(body_field, params.Form)`)
+ or JSON and to generate the JSON Schema for a request body.
+
+ This is **not** used to validate/parse the request body, that's done with each
+ individual body parameter.
+ """
if not flat_dependant.body_params:
return None
first_param = flat_dependant.body_params[0]
- field_info = first_param.field_info
- embed = getattr(field_info, "embed", None)
- body_param_names_set = {param.name for param in flat_dependant.body_params}
- if len(body_param_names_set) == 1 and not embed:
+ if not embed_body_fields:
return first_param
- # If one field requires to embed, all have to be embedded
- # in case a sub-dependency is evaluated with a single unique body field
- # That is combined (embedded) with other body fields
- for param in flat_dependant.body_params:
- setattr(param.field_info, "embed", True) # noqa: B010
model_name = "Body_" + name
BodyModel = create_body_model(
fields=flat_dependant.body_params, model_name=model_name
diff --git a/fastapi/param_functions.py b/fastapi/param_functions.py
index 0d5f27af4..7ddaace25 100644
--- a/fastapi/param_functions.py
+++ b/fastapi/param_functions.py
@@ -1282,7 +1282,7 @@ def Body( # noqa: N802
),
] = _Unset,
embed: Annotated[
- bool,
+ Union[bool, None],
Doc(
"""
When `embed` is `True`, the parameter will be expected in a JSON body as a
@@ -1294,7 +1294,7 @@ def Body( # noqa: N802
[FastAPI docs for Body - Multiple Parameters](https://fastapi.tiangolo.com/tutorial/body-multiple-params/#embed-a-single-body-parameter).
"""
),
- ] = False,
+ ] = None,
media_type: Annotated[
str,
Doc(
diff --git a/fastapi/params.py b/fastapi/params.py
index cc2a5c13c..3dfa5a1a3 100644
--- a/fastapi/params.py
+++ b/fastapi/params.py
@@ -479,7 +479,7 @@ class Body(FieldInfo):
*,
default_factory: Union[Callable[[], Any], None] = _Unset,
annotation: Optional[Any] = None,
- embed: bool = False,
+ embed: Union[bool, None] = None,
media_type: str = "application/json",
alias: Optional[str] = None,
alias_priority: Union[int, None] = _Unset,
@@ -642,7 +642,6 @@ class Form(Body):
default=default,
default_factory=default_factory,
annotation=annotation,
- embed=True,
media_type=media_type,
alias=alias,
alias_priority=alias_priority,
diff --git a/fastapi/routing.py b/fastapi/routing.py
index 2b68cb0d1..7daec9b0f 100644
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@@ -33,8 +33,10 @@ from fastapi._compat import (
from fastapi.datastructures import Default, DefaultPlaceholder
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import (
+ _should_embed_body_fields,
get_body_field,
get_dependant,
+ get_flat_dependant,
get_parameterless_sub_dependant,
get_typed_return_annotation,
solve_dependencies,
@@ -225,6 +227,7 @@ def get_request_handler(
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
dependency_overrides_provider: Optional[Any] = None,
+ embed_body_fields: bool = False,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = asyncio.iscoroutinefunction(dependant.call)
@@ -291,6 +294,7 @@ def get_request_handler(
body=body,
dependency_overrides_provider=dependency_overrides_provider,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
errors = solved_result.errors
if not errors:
@@ -354,7 +358,9 @@ def get_request_handler(
def get_websocket_app(
- dependant: Dependant, dependency_overrides_provider: Optional[Any] = None
+ dependant: Dependant,
+ dependency_overrides_provider: Optional[Any] = None,
+ embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
async def app(websocket: WebSocket) -> None:
async with AsyncExitStack() as async_exit_stack:
@@ -367,6 +373,7 @@ def get_websocket_app(
dependant=dependant,
dependency_overrides_provider=dependency_overrides_provider,
async_exit_stack=async_exit_stack,
+ embed_body_fields=embed_body_fields,
)
if solved_result.errors:
raise WebSocketRequestValidationError(
@@ -399,11 +406,15 @@ class APIWebSocketRoute(routing.WebSocketRoute):
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.app = websocket_session(
get_websocket_app(
dependant=self.dependant,
dependency_overrides_provider=dependency_overrides_provider,
+ embed_body_fields=self._embed_body_fields,
)
)
@@ -544,7 +555,15 @@ class APIRoute(routing.Route):
0,
get_parameterless_sub_dependant(depends=depends, path=self.path_format),
)
- self.body_field = get_body_field(dependant=self.dependant, name=self.unique_id)
+ self._flat_dependant = get_flat_dependant(self.dependant)
+ 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]]:
@@ -561,6 +580,7 @@ class APIRoute(routing.Route):
response_model_exclude_defaults=self.response_model_exclude_defaults,
response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider,
+ embed_body_fields=self._embed_body_fields,
)
def matches(self, scope: Scope) -> Tuple[Match, Scope]:
diff --git a/requirements-tests.txt b/requirements-tests.txt
index 08561d23a..de5fdb8a2 100644
--- a/requirements-tests.txt
+++ b/requirements-tests.txt
@@ -3,7 +3,7 @@
pytest >=7.1.3,<8.0.0
coverage[toml] >= 6.5.0,< 8.0
mypy ==1.8.0
-ruff ==0.6.1
+ruff ==0.6.3
dirty-equals ==0.6.0
# TODO: once removing databases from tutorial, upgrade SQLAlchemy
# probably when including SQLModel
diff --git a/scripts/playwright/request_form_models/image01.py b/scripts/playwright/request_form_models/image01.py
new file mode 100644
index 000000000..15bd3858c
--- /dev/null
+++ b/scripts/playwright/request_form_models/image01.py
@@ -0,0 +1,36 @@
+import subprocess
+import time
+
+import httpx
+from playwright.sync_api import Playwright, sync_playwright
+
+
+# Run playwright codegen to generate the code below, copy paste the sections in run()
+def run(playwright: Playwright) -> None:
+ browser = playwright.chromium.launch(headless=False)
+ context = browser.new_context()
+ page = context.new_page()
+ page.goto("http://localhost:8000/docs")
+ page.get_by_role("button", name="POST /login/ Login").click()
+ page.get_by_role("button", name="Try it out").click()
+ page.screenshot(path="docs/en/docs/img/tutorial/request-form-models/image01.png")
+
+ # ---------------------
+ context.close()
+ browser.close()
+
+
+process = subprocess.Popen(
+ ["fastapi", "run", "docs_src/request_form_models/tutorial001.py"]
+)
+try:
+ for _ in range(3):
+ try:
+ response = httpx.get("http://localhost:8000/docs")
+ except httpx.ConnectError:
+ time.sleep(1)
+ break
+ with sync_playwright() as playwright:
+ run(playwright)
+finally:
+ process.terminate()
diff --git a/tests/test_compat.py b/tests/test_compat.py
index bf268b860..270475bf3 100644
--- a/tests/test_compat.py
+++ b/tests/test_compat.py
@@ -1,11 +1,13 @@
-from typing import List, Union
+from typing import Any, Dict, List, Union
from fastapi import FastAPI, UploadFile
from fastapi._compat import (
ModelField,
Undefined,
_get_model_config,
+ get_model_fields,
is_bytes_sequence_annotation,
+ is_scalar_field,
is_uploadfile_sequence_annotation,
)
from fastapi.testclient import TestClient
@@ -91,3 +93,12 @@ def test_is_uploadfile_sequence_annotation():
# and other types, but I'm not even sure it's a good idea to support it as a first
# class "feature"
assert is_uploadfile_sequence_annotation(Union[List[str], List[UploadFile]])
+
+
+def test_is_pv1_scalar_field():
+ # For coverage
+ class Model(BaseModel):
+ foo: Union[str, Dict[str, Any]]
+
+ fields = get_model_fields(Model)
+ assert not is_scalar_field(fields[0])
diff --git a/tests/test_forms_single_model.py b/tests/test_forms_single_model.py
new file mode 100644
index 000000000..7ed3ba3a2
--- /dev/null
+++ b/tests/test_forms_single_model.py
@@ -0,0 +1,129 @@
+from typing import List, Optional
+
+from dirty_equals import IsDict
+from fastapi import FastAPI, Form
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+class FormModel(BaseModel):
+ username: str
+ lastname: str
+ age: Optional[int] = None
+ tags: List[str] = ["foo", "bar"]
+
+
+@app.post("/form/")
+def post_form(user: Annotated[FormModel, Form()]):
+ return user
+
+
+client = TestClient(app)
+
+
+def test_send_all_data():
+ response = client.post(
+ "/form/",
+ data={
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": "70",
+ "tags": ["plumbus", "citadel"],
+ },
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": 70,
+ "tags": ["plumbus", "citadel"],
+ }
+
+
+def test_defaults():
+ response = client.post("/form/", data={"username": "Rick", "lastname": "Sanchez"})
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": None,
+ "tags": ["foo", "bar"],
+ }
+
+
+def test_invalid_data():
+ response = client.post(
+ "/form/",
+ data={
+ "username": "Rick",
+ "lastname": "Sanchez",
+ "age": "seventy",
+ "tags": ["plumbus", "citadel"],
+ },
+ )
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "int_parsing",
+ "loc": ["body", "age"],
+ "msg": "Input should be a valid integer, unable to parse string as an integer",
+ "input": "seventy",
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "age"],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ }
+ ]
+ }
+ )
+
+
+def test_no_data():
+ response = client.post("/form/")
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"tags": ["foo", "bar"]},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "lastname"],
+ "msg": "Field required",
+ "input": {"tags": ["foo", "bar"]},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "lastname"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
diff --git a/tests/test_forms_single_param.py b/tests/test_forms_single_param.py
new file mode 100644
index 000000000..3bb951441
--- /dev/null
+++ b/tests/test_forms_single_param.py
@@ -0,0 +1,99 @@
+from fastapi import FastAPI, Form
+from fastapi.testclient import TestClient
+from typing_extensions import Annotated
+
+app = FastAPI()
+
+
+@app.post("/form/")
+def post_form(username: Annotated[str, Form()]):
+ return username
+
+
+client = TestClient(app)
+
+
+def test_single_form_field():
+ response = client.post("/form/", data={"username": "Rick"})
+ assert response.status_code == 200, response.text
+ assert response.json() == "Rick"
+
+
+def test_openapi_schema():
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/form/": {
+ "post": {
+ "summary": "Post Form",
+ "operationId": "post_form_form__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {
+ "$ref": "#/components/schemas/Body_post_form_form__post"
+ }
+ }
+ },
+ "required": True,
+ },
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "Body_post_form_form__post": {
+ "properties": {"username": {"type": "string", "title": "Username"}},
+ "type": "object",
+ "required": ["username"],
+ "title": "Body_post_form_form__post",
+ },
+ "HTTPValidationError": {
+ "properties": {
+ "detail": {
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ "type": "array",
+ "title": "Detail",
+ }
+ },
+ "type": "object",
+ "title": "HTTPValidationError",
+ },
+ "ValidationError": {
+ "properties": {
+ "loc": {
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ "type": "array",
+ "title": "Location",
+ },
+ "msg": {"type": "string", "title": "Message"},
+ "type": {"type": "string", "title": "Error Type"},
+ },
+ "type": "object",
+ "required": ["loc", "msg", "type"],
+ "title": "ValidationError",
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/__init__.py b/tests/test_tutorial/test_request_form_models/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001.py b/tests/test_tutorial/test_request_form_models/test_tutorial001.py
new file mode 100644
index 000000000..46c130ee8
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001.py
@@ -0,0 +1,232 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001 import app
+
+ client = TestClient(app)
+ return client
+
+
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py b/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py
new file mode 100644
index 000000000..4e14d89c8
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001_an.py
@@ -0,0 +1,232 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001_an import app
+
+ client = TestClient(app)
+ return client
+
+
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }
diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py b/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py
new file mode 100644
index 000000000..2e6426aa7
--- /dev/null
+++ b/tests/test_tutorial/test_request_form_models/test_tutorial001_an_py39.py
@@ -0,0 +1,240 @@
+import pytest
+from dirty_equals import IsDict
+from fastapi.testclient import TestClient
+
+from tests.utils import needs_py39
+
+
+@pytest.fixture(name="client")
+def get_client():
+ from docs_src.request_form_models.tutorial001_an_py39 import app
+
+ client = TestClient(app)
+ return client
+
+
+@needs_py39
+def test_post_body_form(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo", "password": "secret"})
+ assert response.status_code == 200
+ assert response.json() == {"username": "Foo", "password": "secret"}
+
+
+@needs_py39
+def test_post_body_form_no_password(client: TestClient):
+ response = client.post("/login/", data={"username": "Foo"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {"username": "Foo"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_form_no_username(client: TestClient):
+ response = client.post("/login/", data={"password": "secret"})
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {"password": "secret"},
+ }
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_form_no_data(client: TestClient):
+ response = client.post("/login/")
+ assert response.status_code == 422
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_post_body_json(client: TestClient):
+ response = client.post("/login/", json={"username": "Foo", "password": "secret"})
+ assert response.status_code == 422, response.text
+ assert response.json() == IsDict(
+ {
+ "detail": [
+ {
+ "type": "missing",
+ "loc": ["body", "username"],
+ "msg": "Field required",
+ "input": {},
+ },
+ {
+ "type": "missing",
+ "loc": ["body", "password"],
+ "msg": "Field required",
+ "input": {},
+ },
+ ]
+ }
+ ) | IsDict(
+ # TODO: remove when deprecating Pydantic v1
+ {
+ "detail": [
+ {
+ "loc": ["body", "username"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ {
+ "loc": ["body", "password"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ },
+ ]
+ }
+ )
+
+
+@needs_py39
+def test_openapi_schema(client: TestClient):
+ response = client.get("/openapi.json")
+ assert response.status_code == 200, response.text
+ assert response.json() == {
+ "openapi": "3.1.0",
+ "info": {"title": "FastAPI", "version": "0.1.0"},
+ "paths": {
+ "/login/": {
+ "post": {
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "summary": "Login",
+ "operationId": "login_login__post",
+ "requestBody": {
+ "content": {
+ "application/x-www-form-urlencoded": {
+ "schema": {"$ref": "#/components/schemas/FormData"}
+ }
+ },
+ "required": True,
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "FormData": {
+ "properties": {
+ "username": {"type": "string", "title": "Username"},
+ "password": {"type": "string", "title": "Password"},
+ },
+ "type": "object",
+ "required": ["username", "password"],
+ "title": "FormData",
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {
+ "anyOf": [{"type": "string"}, {"type": "integer"}]
+ },
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ }
+ },
+ }