From 48e43e5cbf82415cd17ec542e67058a9be4db48e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Wed, 21 Oct 2026 18:43:09 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20support=20for=20combining=20f?= =?UTF-8?q?orm=20models=20(#16577)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/en/docs/tutorial/request-form-models.md | 18 + .../tutorial003_an_py310.py | 23 + .../request_form_models/tutorial003_py310.py | 21 + .../tutorial004_an_py310.py | 15 + .../request_form_models/tutorial004_py310.py | 13 + fastapi/_compat/v2.py | 6 +- fastapi/dependencies/utils.py | 142 +++++- tests/test_forms_multiple_models.py | 439 ++++++++++++++++++ .../test_tutorial003.py | 33 ++ .../test_tutorial004.py | 40 ++ 10 files changed, 726 insertions(+), 24 deletions(-) create mode 100644 docs_src/request_form_models/tutorial003_an_py310.py create mode 100644 docs_src/request_form_models/tutorial003_py310.py create mode 100644 docs_src/request_form_models/tutorial004_an_py310.py create mode 100644 docs_src/request_form_models/tutorial004_py310.py create mode 100644 tests/test_forms_multiple_models.py create mode 100644 tests/test_tutorial/test_request_form_models/test_tutorial003.py create mode 100644 tests/test_tutorial/test_request_form_models/test_tutorial004.py diff --git a/docs/en/docs/tutorial/request-form-models.md b/docs/en/docs/tutorial/request-form-models.md index fea03e0ff4..7e023400d4 100644 --- a/docs/en/docs/tutorial/request-form-models.md +++ b/docs/en/docs/tutorial/request-form-models.md @@ -75,6 +75,24 @@ They will receive an error response telling them that the field `extra` is not a } ``` +## Combine Form Models { #combine-form-models } + +You can combine multiple form models with individual `Form` parameters. + +{* ../../docs_src/request_form_models/tutorial003_an_py310.py hl[9:14,19:21] *} + +The client sends `name`, `tags`, and `token` as form fields. Each model receives its own fields and validates them independently. + +If multiple declarations use the same form field name, each validates that field's value. Use compatible types and constraints for shared fields so the OpenAPI schema can describe them consistently. + +## Form Models and Files { #form-models-and-files } + +You can also combine a form model with [file parameters](request-files.md). + +{* ../../docs_src/request_form_models/tutorial004_an_py310.py hl[14] *} + +Here the client sends the `name` field and the `avatar` file using `multipart/form-data`. + ## Summary { #summary } You can use Pydantic models to declare form fields in FastAPI. 😎 diff --git a/docs_src/request_form_models/tutorial003_an_py310.py b/docs_src/request_form_models/tutorial003_an_py310.py new file mode 100644 index 0000000000..14aea2532d --- /dev/null +++ b/docs_src/request_form_models/tutorial003_an_py310.py @@ -0,0 +1,23 @@ +from typing import Annotated + +from fastapi import FastAPI, Form +from pydantic import BaseModel + +app = FastAPI() + + +class Profile(BaseModel): + name: str + + +class Preferences(BaseModel): + tags: list[str] = [] + + +@app.post("/profiles/") +async def create_profile( + profile: Annotated[Profile, Form()], + preferences: Annotated[Preferences, Form()], + token: Annotated[str, Form()], +): + return {"profile": profile, "preferences": preferences, "token": token} diff --git a/docs_src/request_form_models/tutorial003_py310.py b/docs_src/request_form_models/tutorial003_py310.py new file mode 100644 index 0000000000..9ca41eaaa2 --- /dev/null +++ b/docs_src/request_form_models/tutorial003_py310.py @@ -0,0 +1,21 @@ +from fastapi import FastAPI, Form +from pydantic import BaseModel + +app = FastAPI() + + +class Profile(BaseModel): + name: str + + +class Preferences(BaseModel): + tags: list[str] = [] + + +@app.post("/profiles/") +async def create_profile( + profile: Profile = Form(), + preferences: Preferences = Form(), + token: str = Form(), +): + return {"profile": profile, "preferences": preferences, "token": token} diff --git a/docs_src/request_form_models/tutorial004_an_py310.py b/docs_src/request_form_models/tutorial004_an_py310.py new file mode 100644 index 0000000000..4de846f4d6 --- /dev/null +++ b/docs_src/request_form_models/tutorial004_an_py310.py @@ -0,0 +1,15 @@ +from typing import Annotated + +from fastapi import FastAPI, Form, UploadFile +from pydantic import BaseModel + +app = FastAPI() + + +class Profile(BaseModel): + name: str + + +@app.post("/profiles/") +async def create_profile(profile: Annotated[Profile, Form()], avatar: UploadFile): + return {"profile": profile, "filename": avatar.filename} diff --git a/docs_src/request_form_models/tutorial004_py310.py b/docs_src/request_form_models/tutorial004_py310.py new file mode 100644 index 0000000000..d712c1fc0a --- /dev/null +++ b/docs_src/request_form_models/tutorial004_py310.py @@ -0,0 +1,13 @@ +from fastapi import FastAPI, File, Form, UploadFile +from pydantic import BaseModel + +app = FastAPI() + + +class Profile(BaseModel): + name: str + + +@app.post("/profiles/") +async def create_profile(profile: Profile = Form(), avatar: UploadFile = File()): + return {"profile": profile, "filename": avatar.filename} diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index 8f5cff8911..640e862284 100644 --- a/fastapi/_compat/v2.py +++ b/fastapi/_compat/v2.py @@ -390,10 +390,12 @@ def get_missing_field_error(loc: tuple[int | str, ...]) -> dict[str, Any]: def create_body_model( - *, fields: Sequence[ModelField], model_name: str + *, fields: Sequence[ModelField], model_name: str, config: ConfigDict | None = None ) -> type[BaseModel]: field_params = {f.name: (f.field_info.annotation, f.field_info) for f in fields} - BodyModel: type[BaseModel] = create_model(model_name, **field_params) # type: ignore[call-overload] # ty: ignore[no-matching-overload] + BodyModel: type[BaseModel] = create_model( + model_name, __config__=config, **field_params + ) # type: ignore[call-overload] # ty: ignore[no-matching-overload] return BodyModel diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 5543231cfa..3878d71bf6 100644 --- a/fastapi/dependencies/utils.py +++ b/fastapi/dependencies/utils.py @@ -70,7 +70,15 @@ from fastapi.logger import logger from fastapi.security.oauth2 import SecurityScopes from fastapi.types import DependencyCacheKey from fastapi.utils import create_model_field, get_path_param_names -from pydantic import AliasChoices, AliasPath, BaseModel, BeforeValidator, Json, Tag +from pydantic import ( + AliasChoices, + AliasPath, + BaseModel, + BeforeValidator, + ConfigDict, + Json, + Tag, +) from pydantic.fields import FieldInfo from starlette.background import BackgroundTasks as StarletteBackgroundTasks from starlette.concurrency import run_in_threadpool @@ -171,7 +179,12 @@ def _get_request_param_names( dependant: Dependant, dependency_overrides_provider: Any | None = None ) -> dict[str, set[str]]: """Collect accepted names across the effective dependency graph, by location.""" - names: dict[str, set[str]] = {"query": set(), "header": set(), "cookie": set()} + names: dict[str, set[str]] = { + "query": set(), + "header": set(), + "cookie": set(), + "form": set(), + } overrides = ( getattr(dependency_overrides_provider, "dependency_overrides", None) or {} ) @@ -182,9 +195,22 @@ def _get_request_param_names( ("query", current.query_params), ("header", current.header_params), ("cookie", current.cookie_params), + ( + "form", + [ + f + for f in current.body_params + if isinstance(f.field_info, params.Form) + ], + ), ): for field in fields: - for param in _get_flat_fields_from_params([field]): + model_fields = ( + [field] + if location == "form" and not _is_form_model(field) + else _get_flat_fields_from_params([field]) + ) + for param in model_fields: convert_underscores = location == "header" and getattr( param.field_info, "convert_underscores", @@ -791,6 +817,7 @@ async def solve_dependencies( body_fields=dependant.body_params, received_body=body, embed_body_fields=embed_body_fields, + known_param_names=_request_param_names["form"], ) values.update(body_values) errors.extend(body_errors) @@ -1051,6 +1078,15 @@ def _normalize_form_model(value: Any, *, model: type[BaseModel]) -> Any: return value +def _is_form_model(field: ModelField) -> bool: + return ( + isinstance(field.field_info, params.Form) + and not field.field_info.embed + and lenient_issubclass(field.field_info.annotation, BaseModel) + and not _is_json_field(field) + ) + + def _should_embed_body_fields(fields: list[ModelField]) -> bool: if not fields: return False @@ -1061,6 +1097,13 @@ def _should_embed_body_fields(fields: list[ModelField]) -> bool: if len(body_param_names_set) > 1: return True first_field = fields[0] + # Dependencies can use the same parameter name for different form models. + if _is_form_model(first_field) and any( + not _is_form_model(field) + or field.field_info.annotation is not first_field.field_info.annotation + for field in fields[1:] + ): + return True # If it explicitly specifies it is embedded, it has to be embedded if getattr(first_field.field_info, "embed", None): return True @@ -1078,11 +1121,14 @@ def _should_embed_body_fields(fields: list[ModelField]) -> bool: def _extract_form_items( body_fields: list[ModelField], received_body: FormData, + known_param_names: set[str] | None = None, ) -> Iterator[tuple[str, Any, ModelField | None]]: field_aliases: set[str] = set() + own_names: set[str] = set() + if known_param_names is None: + known_param_names = set() for field in body_fields: - aliases = [get_validation_alias(field)] validation_alias = field.field_info.validation_alias has_alias_path = isinstance(validation_alias, AliasPath) or ( isinstance(validation_alias, AliasChoices) @@ -1090,17 +1136,8 @@ def _extract_form_items( isinstance(choice, AliasPath) for choice in validation_alias.choices ) ) - if not has_alias_path: - if isinstance(validation_alias, AliasChoices): - aliases = list(cast(list[str], validation_alias.choices)) - config = field.config or {} - if config.get("validate_by_alias") is False: - aliases = [field.name] - elif ( - config.get("validate_by_name", config.get("populate_by_name", False)) - and field.name not in aliases - ): - aliases.append(field.name) + aliases = [alias for alias, _ in _get_param_aliases(field)] + own_names.update(aliases) # Interpret missing boolean form fields as unchecked HTML checkboxes. infer_false_if_missing = ( not has_alias_path @@ -1118,7 +1155,9 @@ def _extract_form_items( # Leave later aliases unchanged for other fields or extra validation. break for key in received_body.keys(): - if key not in field_aliases: + if key not in field_aliases and ( + key not in known_param_names or key in own_names + ): param_values = received_body.getlist(key) if len(param_values) == 1: yield key, param_values[0], None @@ -1138,8 +1177,11 @@ async def _read_file_bytes(file: UploadFile) -> bytes: async def _extract_form_body( body_fields: list[ModelField], received_body: FormData, + known_param_names: set[str] | None = None, ) -> AsyncIterator[tuple[str, Any, ModelField | None]]: - for alias, value, field in _extract_form_items(body_fields, received_body): + for alias, value, field in _extract_form_items( + body_fields, received_body, known_param_names + ): if field is not None and isinstance(field.field_info, params.File): if is_bytes_or_nonable_bytes_annotation( field.field_info.annotation @@ -1163,16 +1205,21 @@ async def request_body_to_args( body_fields: list[ModelField], received_body: Any, embed_body_fields: bool, + known_param_names: set[str] | None = None, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: values: dict[str, Any] = {} errors: list[dict[str, Any]] = [] 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 + single_not_embedded_field = ( + len(body_fields) == 1 + and not embed_body_fields + and not _should_embed_body_fields(body_fields) + ) first_field = body_fields[0] body_to_process = received_body form_values: dict[str, tuple[str, Any]] = {} - fields_to_extract: list[ModelField] = body_fields + fields_to_extract = [field for field in body_fields if not _is_form_model(field)] if ( single_not_embedded_field @@ -1190,7 +1237,11 @@ async def request_body_to_args( fields_to_extract = [] body_to_process = {} async for alias, value, field in _extract_form_body( - fields_to_extract, received_body + fields_to_extract, + received_body, + known_param_names + if single_not_embedded_field and _is_form_model(first_field) + else None, ): if single_not_embedded_field: body_to_process[alias] = value @@ -1208,7 +1259,20 @@ async def request_body_to_args( aliases = [alias for alias, _ in _get_param_aliases(field)] loc = ("body", aliases[0]) value = Undefined - if isinstance(received_body, FormData): + if isinstance(received_body, FormData) and _is_form_model(field): + # Extract each model independently from the original form values. + value = { + alias: item + async for alias, item, _ in _extract_form_body( + get_cached_model_fields( + cast(type[BaseModel], field.field_info.annotation) + ), + received_body, + known_param_names, + ) + } + loc = ("body",) + elif isinstance(received_body, FormData): alias, value = form_values.get(field.name, (aliases[0], Undefined)) loc = ("body", alias) elif body_to_process is not Undefined and not isinstance( @@ -1253,7 +1317,41 @@ def _get_body_field( if not embed_body_fields: return first_param model_name = "Body_" + name - BodyModel = create_body_model(fields=body_params, model_name=model_name) + fields = body_params + config: ConfigDict | None = None + if any(_is_form_model(field) for field in body_params): + fields_by_alias: dict[str, ModelField] = {} + config = {} + for field in body_params: + model_fields = [field] + if _is_form_model(field): + model = cast(type[BaseModel], field.field_info.annotation) + model_fields = get_cached_model_fields(model) + extra = model.model_config.get("extra") + if extra in ("allow", "forbid") and config.get("extra") != "forbid": + config["extra"] = extra + for model_field in model_fields: + alias = _get_param_aliases(model_field)[0][0] + previous = fields_by_alias.get(alias) + if previous is not None and ( + previous.field_info.is_required() + or not model_field.field_info.is_required() + ): + continue + # Unique internal names keep identically named fields from different + # models. Copies leave cached fields and their aliases unchanged. + field_info = copy(model_field.field_info) + field_info.alias = alias + field_info.validation_alias = alias + field_info.serialization_alias = alias + fields_by_alias[alias] = ModelField( + name=previous.name + if previous is not None + else f"field_{len(fields_by_alias)}", + field_info=field_info, + ) + fields = list(fields_by_alias.values()) + BodyModel = create_body_model(fields=fields, model_name=model_name, config=config) required = any(True for f in body_params if f.field_info.is_required()) BodyFieldInfo_kwargs: dict[str, Any] = { "annotation": BodyModel, diff --git a/tests/test_forms_multiple_models.py b/tests/test_forms_multiple_models.py new file mode 100644 index 0000000000..b1d24cb5f3 --- /dev/null +++ b/tests/test_forms_multiple_models.py @@ -0,0 +1,439 @@ +from typing import Annotated +from urllib.parse import urlencode + +import pytest +from fastapi import APIRouter, Depends, FastAPI, File, Form, Query, UploadFile +from fastapi.testclient import TestClient +from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator + + +def form_schema(app, path="/", media_type="application/x-www-form-urlencoded"): + schema = app.openapi() + body = schema["paths"][path]["post"]["requestBody"] + ref = body["content"][media_type]["schema"]["$ref"] + return schema["components"]["schemas"][ref.rsplit("/", 1)[-1]] + + +@pytest.mark.parametrize("multipart", [False, True]) +@pytest.mark.parametrize("in_dependency", [False, True]) +@pytest.mark.parametrize("extra", ["ignore", "allow", "forbid"]) +def test_composed_form_models(multipart, in_dependency, extra): + app = FastAPI() + + class Profile(BaseModel): + model_config = ConfigDict(extra=extra) + name: str + + class Preferences(BaseModel): + model_config = ConfigDict(extra=extra) + limit: int = Field(10, ge=1, description="Result limit") + tags: list[str] = Field(default_factory=list) + + def profile_dependency(profile: Annotated[Profile, Form()]): + return profile + + marker = Depends(profile_dependency) if in_dependency else Form() + + @app.post("/") + def submit( + profile: Annotated[Profile, marker], + preferences: Annotated[Preferences, Form()], + token: Annotated[str, Form()], + ): + return {"profile": profile, "preferences": preferences, "token": token} + + client = TestClient(app) + + def send(fields): + if multipart: + return client.post("/", files=[(k, (None, v)) for k, v in fields]) + return client.post( + "/", + content=urlencode(fields), + headers={"content-type": "application/x-www-form-urlencoded"}, + ) + + fields = [("name", "Rick"), ("token", "secret")] + response = send(fields + [("limit", "2"), ("tags", "a"), ("tags", "b")]) + assert response.status_code == 200, response.text + assert response.json() == { + "profile": {"name": "Rick"}, + "preferences": {"limit": 2, "tags": ["a", "b"]}, + "token": "secret", + } + response = send(fields) + assert response.status_code == 200, response.text + assert response.json()["preferences"] == {"limit": 10, "tags": []} + response = send(fields + [("limit", "0")]) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "limit"] + response = send(fields + [("unknown", "a"), ("unknown", "b")]) + if extra == "forbid": + assert response.status_code == 422 + assert [error["loc"] for error in response.json()["detail"]] == [ + ["body", "unknown"], + ["body", "unknown"], + ] + else: + assert response.status_code == 200, response.text + for key in ("profile", "preferences"): + if extra == "allow": + assert response.json()[key]["unknown"] == ["a", "b"] + else: + assert "unknown" not in response.json()[key] + response = send([("token", "secret")]) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "name"] + schema = form_schema(app) + assert set(schema["properties"]) == {"name", "limit", "tags", "token"} + assert set(schema["required"]) == {"name", "token"} + assert schema["properties"]["limit"] == { + "type": "integer", + "minimum": 1, + "title": "Limit", + "description": "Result limit", + "default": 10, + } + if extra == "forbid": + assert schema["additionalProperties"] is False + elif extra == "allow": + assert schema["additionalProperties"] is True + + +def test_model_defaults_checkboxes_and_validators_are_independent(): + app = FastAPI() + + class Settings(BaseModel): + label: str = "default" + enabled: bool = True + tags: list[str] = Field(default_factory=list) + + @model_validator(mode="before") + @classmethod + def check_input(cls, value): + assert "token" not in value + return value + + @model_validator(mode="after") + def check_label(self): + if self.label == "invalid": + raise ValueError("Invalid label") + return self + + class OtherSettings(BaseModel): + label: str = "other" + enabled: str = "other" + + @app.post("/") + def submit( + settings: Annotated[Settings, Form()], + other: Annotated[OtherSettings, Form()], + token: Annotated[str, Form()] = "secret", + ): + return { + "settings": settings, + "other": other, + "fields_set": sorted(settings.model_fields_set), + "other_fields_set": sorted(other.model_fields_set), + } + + client = TestClient(app) + response = client.post("/") + assert response.status_code == 200, response.text + assert response.json() == { + "settings": {"label": "default", "enabled": False, "tags": []}, + "other": {"label": "other", "enabled": "other"}, + "fields_set": ["enabled"], + "other_fields_set": [], + } + response = client.post( + "/", data={"label": "", "enabled": "", "tags": ["", "a"], "token": "ok"} + ) + assert response.status_code == 200, response.text + assert response.json() == { + "settings": {"label": "default", "enabled": True, "tags": ["", "a"]}, + "other": {"label": "other", "enabled": "other"}, + "fields_set": ["tags"], + "other_fields_set": [], + } + response = client.post("/", data={"label": "invalid"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body"] + + +@pytest.mark.parametrize("required_first", [False, True]) +def test_aliases_shared_fields_and_cached_models(required_first): + app = FastAPI() + + class First(BaseModel): + model_config = ConfigDict(extra="forbid", populate_by_name=True) + value: int = Field(alias="first", description="First value") + shared: int = Field() if required_first else Field(10) + + class Second(BaseModel): + value: int = Field(alias="second") + shared: int = Field(10) if required_first else Field() + + @app.post("/") + def submit(first: Annotated[First, Form()], second: Annotated[Second, Form()]): + return [first, second] + + @app.post("/single") + def single(data: Annotated[First, Form()]): + return data + + @app.post("/json") + def json_body(first: First, second: Second): + return [first, second] + + client = TestClient(app) + response = client.post("/", data={"value": "1", "second": "2", "shared": "3"}) + assert response.status_code == 200, response.text + expected = [{"first": 1, "shared": 3}, {"second": 2, "shared": 3}] + assert response.json() == expected + assert client.post("/", data={"first": "1", "second": "2"}).status_code == 422 + schema = form_schema(app) + assert set(schema["properties"]) == {"first", "second", "shared"} + assert set(schema["required"]) == {"first", "second", "shared"} + assert "default" not in schema["properties"]["shared"] + assert schema["additionalProperties"] is False + assert set(form_schema(app, "/single")["properties"]) == {"first", "shared"} + assert set(form_schema(app, "/json", "application/json")["properties"]) == { + "first", + "second", + } + assert First.model_fields["value"].alias == "first" + assert First.model_fields["shared"].alias is None + assert First.model_fields["shared"].is_required() is required_first + response = client.post("/single", data={"first": "4", "shared": "5"}) + assert response.status_code == 200, response.text + assert response.json() == {"first": 4, "shared": 5} + response = client.post( + "/json", + json={"first": {"first": 1, "shared": 3}, "second": {"second": 2, "shared": 3}}, + ) + assert response.status_code == 200, response.text + + +def test_shared_alias_choices_keep_shapes_and_unused_extras(): + app = FastAPI() + + class Many(BaseModel): + model_config = ConfigDict(extra="forbid") + names: list[str] = Field(validation_alias=AliasChoices("names", "legacy")) + + class One(BaseModel): + text: str = Field(alias="legacy") + + @app.post("/") + def submit( + many: Annotated[Many, Form()], + one: Annotated[One, Form()], + legacy: Annotated[str, Form()], + ): + return {"many": many, "one": one, "last": legacy} + + client = TestClient(app) + response = client.post("/", data={"legacy": ["first", "last"]}) + assert response.status_code == 200, response.text + assert response.json() == { + "many": {"names": ["first", "last"]}, + "one": {"legacy": "last"}, + "last": "last", + } + response = client.post("/", data={"names": "first", "legacy": "last"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "legacy"] + + +def test_nested_dependencies_overrides_and_request_locations(): + app = FastAPI() + + class Profile(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str + + class Preferences(BaseModel): + model_config = ConfigDict(extra="forbid") + limit: int + + def preferences(data: Annotated[Preferences, Form()]): + return data.limit + + def replacement(page: Annotated[int, Form()] = 1): + return page + + def nested(value: Annotated[int, Depends(preferences)]): + return value + + def token(token: Annotated[str, Form()]): + return token + + router = APIRouter(dependencies=[Depends(token)]) + + @router.post("/") + def submit( + data: Annotated[Profile, Form()], + value: Annotated[int, Depends(nested)], + query: Annotated[str, Query()] = "default", + ): + return {"data": data, "value": value} + + app.include_router(router) + client = TestClient(app) + fields = {"name": "Rick", "token": "secret"} + response = client.post("/?query=ok", data={**fields, "limit": "2"}) + assert response.status_code == 200, response.text + assert response.json() == {"data": {"name": "Rick"}, "value": 2} + assert set(form_schema(app)["properties"]) == {"name", "limit", "token"} + app.dependency_overrides[preferences] = replacement + response = client.post("/", data={**fields, "page": "3"}) + assert response.status_code == 200, response.text + assert response.json()["value"] == 3 + for unknown in ("limit", "query"): + response = client.post("/", data={**fields, unknown: "2"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", unknown] + app.dependency_overrides.clear() + assert client.post("/", data={**fields, "limit": "4"}).status_code == 200 + + +@pytest.mark.parametrize("content", [b"", b"hello"]) +def test_form_model_and_files_shared_with_dependencies(content): + app = FastAPI() + + class Profile(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str + + def inspect_files( + profile: Annotated[Profile, Form()], files: Annotated[list[bytes], File()] + ): + return {"profile": profile, "files": [file.decode() for file in files]} + + @app.post("/") + async def submit( + inspected: Annotated[dict, Depends(inspect_files)], + profile: Annotated[Profile, Form()], + files: Annotated[list[UploadFile], File()], + last: Annotated[bytes, File(alias="files")], + ): + return { + "inspected": inspected, + "profile": profile, + "positions": [file.file.tell() for file in files], + "files": [(await file.read()).decode() for file in files], + "last": last.decode(), + } + + client = TestClient(app) + response = client.post( + "/", + data={"name": "Rick"}, + files=[("files", ("one.txt", b"one")), ("files", ("two.txt", content))], + ) + assert response.status_code == 200, response.text + expected_files = ["one", content.decode()] + assert response.json() == { + "inspected": {"profile": {"name": "Rick"}, "files": expected_files}, + "profile": {"name": "Rick"}, + "positions": [0, 0], + "files": expected_files, + "last": content.decode(), + } + response = client.post("/", data={"name": "Rick"}) + assert response.status_code == 422 + assert all(error["loc"] == ["body", "files"] for error in response.json()["detail"]) + schema = form_schema(app, media_type="multipart/form-data") + assert set(schema["properties"]) == {"name", "files"} + assert set(schema["required"]) == {"name", "files"} + assert schema["properties"]["files"]["type"] == "array" + assert schema["additionalProperties"] is False + + +def test_models_with_the_same_parameter_name_in_dependencies(): + app = FastAPI() + + class First(BaseModel): + model_config = ConfigDict(extra="forbid") + first: int + + class Second(BaseModel): + model_config = ConfigDict(extra="forbid") + second: int + + def dependency(data: Annotated[First, Form()]): + return data + + @app.post("/") + def submit( + data: Annotated[Second, Form()], first: Annotated[First, Depends(dependency)] + ): + return [first, data] + + response = TestClient(app).post("/", data={"first": "1", "second": "2"}) + assert response.status_code == 200, response.text + assert response.json() == [{"first": 1}, {"second": 2}] + assert set(form_schema(app)["properties"]) == {"first", "second"} + + +def test_optional_file_with_a_form_model(): + app = FastAPI() + + class Profile(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str + + @app.post("/") + def submit( + profile: Annotated[Profile, Form()], + attachment: Annotated[bytes | None, File()] = None, + ): + return {"profile": profile, "attachment": attachment} + + client = TestClient(app) + response = client.post("/", data={"name": "Rick"}) + assert response.status_code == 200, response.text + assert response.json() == {"profile": {"name": "Rick"}, "attachment": None} + response = client.post( + "/", data={"name": "Rick"}, files={"attachment": ("empty.txt", b"")} + ) + assert response.status_code == 200, response.text + assert response.json() == {"profile": {"name": "Rick"}, "attachment": ""} + assert form_schema(app, media_type="multipart/form-data")["required"] == ["name"] + + +def test_dependency_override_adds_a_form_parameter(): + app = FastAPI() + + class Profile(BaseModel): + model_config = ConfigDict(extra="forbid") + name: str + + def token(): + return "default" + + def replacement(token: Annotated[str, Form()]): + return token + + @app.post("/") + def submit( + profile: Annotated[Profile, Form()], value: Annotated[str, Depends(token)] + ): + return {"profile": profile, "token": value} + + client = TestClient(app) + app.dependency_overrides[token] = replacement + response = client.post("/", data={"name": "Rick", "token": "secret"}) + assert response.status_code == 200, response.text + assert response.json() == {"profile": {"name": "Rick"}, "token": "secret"} + response = client.post("/", data={"name": "Rick"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "token"] + app.dependency_overrides.clear() + response = client.post("/", data={"name": "Rick", "token": "secret"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "token"] + response = client.post("/", data={"name": "Rick"}) + assert response.status_code == 200, response.text + assert response.json() == {"profile": {"name": "Rick"}, "token": "default"} diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial003.py b/tests/test_tutorial/test_request_form_models/test_tutorial003.py new file mode 100644 index 0000000000..de552861ff --- /dev/null +++ b/tests/test_tutorial/test_request_form_models/test_tutorial003.py @@ -0,0 +1,33 @@ +import importlib + +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture(name="client", params=["tutorial003_py310", "tutorial003_an_py310"]) +def get_client(request: pytest.FixtureRequest): + mod = importlib.import_module(f"docs_src.request_form_models.{request.param}") + return TestClient(mod.app) + + +def test_form_models(client: TestClient): + response = client.post( + "/profiles/", + data={"name": "Rick", "tags": ["python", "fastapi"], "token": "secret"}, + ) + assert response.status_code == 200, response.text + assert response.json() == { + "profile": {"name": "Rick"}, + "preferences": {"tags": ["python", "fastapi"]}, + "token": "secret", + } + response = client.post("/profiles/", data={"name": "Rick", "token": "secret"}) + assert response.status_code == 200, response.text + assert response.json()["preferences"] == {"tags": []} + + +def test_openapi(client: TestClient): + schema = client.get("/openapi.json").json() + body = schema["components"]["schemas"]["Body_create_profile_profiles__post"] + assert set(body["properties"]) == {"name", "tags", "token"} + assert set(body["required"]) == {"name", "token"} diff --git a/tests/test_tutorial/test_request_form_models/test_tutorial004.py b/tests/test_tutorial/test_request_form_models/test_tutorial004.py new file mode 100644 index 0000000000..8d3dd2e112 --- /dev/null +++ b/tests/test_tutorial/test_request_form_models/test_tutorial004.py @@ -0,0 +1,40 @@ +import importlib + +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture(name="client", params=["tutorial004_py310", "tutorial004_an_py310"]) +def get_client(request: pytest.FixtureRequest): + mod = importlib.import_module(f"docs_src.request_form_models.{request.param}") + return TestClient(mod.app) + + +def test_form_model_and_file(client: TestClient): + response = client.post( + "/profiles/", data={"name": "Rick"}, files={"avatar": ("avatar.png", b"image")} + ) + assert response.status_code == 200, response.text + assert response.json() == {"profile": {"name": "Rick"}, "filename": "avatar.png"} + + +def test_missing_file(client: TestClient): + response = client.post("/profiles/", data={"name": "Rick"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "avatar"] + + +def test_openapi(client: TestClient): + schema = client.get("/openapi.json").json() + body = schema["paths"]["/profiles/"]["post"]["requestBody"] + assert set(body["content"]) == {"multipart/form-data"} + model = schema["components"]["schemas"]["Body_create_profile_profiles__post"] + assert model["properties"] == { + "name": {"type": "string", "title": "Name"}, + "avatar": { + "type": "string", + "contentMediaType": "application/octet-stream", + "title": "Avatar", + }, + } + assert set(model["required"]) == {"name", "avatar"}