From dfb3b91b7c6102cddde6e2cb1422d0faaa273be5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Tue, 20 Oct 2026 23:37:24 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Fix=20`AliasChoices`=20for=20que?= =?UTF-8?q?ry=20parameters,=20headers,=20and=20cookies=20(#16555)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fastapi/_compat/v2.py | 14 +- fastapi/dependencies/utils.py | 83 ++- fastapi/openapi/utils.py | 21 +- tests/test_header_cookie_alias_choices.py | 649 ++++++++++++++++++++++ tests/test_query_alias_choices.py | 466 ++++++++++++++++ 5 files changed, 1191 insertions(+), 42 deletions(-) create mode 100644 tests/test_header_cookie_alias_choices.py create mode 100644 tests/test_query_alias_choices.py diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index f31878e75f..8f5cff8911 100644 --- a/fastapi/_compat/v2.py +++ b/fastapi/_compat/v2.py @@ -259,20 +259,22 @@ def get_schema_from_model_field( tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue ], separate_input_output_schemas: bool = True, + field_alias: str | None = None, ) -> dict[str, Any]: override_mode: Literal["validation"] | None = ( None if (separate_input_output_schemas or _has_computed_fields(field)) else "validation" ) - field_alias = ( - (field.validation_alias or field.alias) - if field.mode == "validation" - else (field.serialization_alias or field.alias) - ) + if field_alias is None: + field_alias = ( + (field.validation_alias or field.alias) + if field.mode == "validation" + else (field.serialization_alias or field.alias) + ) # This expects that GenerateJsonSchema was already used to generate the definitions - json_schema = field_mapping[(field, override_mode or field.mode)] + json_schema = field_mapping[(field, override_mode or field.mode)].copy() if "$ref" not in json_schema: # TODO remove when deprecating Pydantic v1 # Ref: https://github.com/pydantic/pydantic/blob/d61792cc42c80b13b23e3ffa74bc37ec7c77f7d1/pydantic/schema.py#L207 diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 36702b332f..f5f7fe9ef3 100644 --- a/fastapi/dependencies/utils.py +++ b/fastapi/dependencies/utils.py @@ -769,9 +769,8 @@ def _is_empty_file_input(value: Any) -> bool: def _get_multidict_value( - field: ModelField, values: Mapping[str, Any], alias: str | None = None + field: ModelField, values: Mapping[str, Any], alias: str ) -> Any: - alias = alias or get_validation_alias(field) if ( (not _is_json_field(field)) and field_annotation_is_sequence(field.field_info.annotation) @@ -831,26 +830,28 @@ def request_params_to_args( processed_keys = set() for field in fields_to_extract: - alias = None - if isinstance(received_params, Headers): - # Handle fields extracted from a Pydantic Model for a header, each field - # doesn't have a FieldInfo of type Header with the default convert_underscores=True - convert_underscores = getattr( - field.field_info, "convert_underscores", default_convert_underscores - ) - if convert_underscores: - alias = get_validation_alias(field) - if alias == field.name: - alias = alias.replace("_", "-") - header_aliases[field.name] = alias - value = _get_multidict_value(field, received_params, alias=alias) + convert_underscores = isinstance(received_params, Headers) and getattr( + field.field_info, "convert_underscores", default_convert_underscores + ) + aliases = _get_param_aliases(field, convert_underscores=convert_underscores) + alias, request_alias = next( + (pair for pair in aliases if pair[1] in received_params), aliases[0] + ) + # Only reshape the selected alias. Unused choices can be extra fields + # or aliases used by other fields with different types. + value = _get_multidict_value(field, received_params, alias=request_alias) if value is not Undefined: - params_to_process[get_validation_alias(field)] = value - processed_keys.add(alias or get_validation_alias(field)) - # For headers with convert_underscores=True, mark both the converted - # header name and the original field alias as processed to avoid - # accepting the original alias as an extra header. - processed_keys.add(get_validation_alias(field)) + params_to_process[alias] = value + if isinstance(received_params, Headers): + processed_keys.add(request_alias.lower()) + for validation_alias, header_alias in aliases: + if validation_alias != header_alias: + # Do not accept an unconverted field name as an extra header. + processed_keys.add(validation_alias.lower()) + if alias != request_alias: + header_aliases[field.name] = request_alias + else: + processed_keys.add(request_alias) for key in received_params.keys(): if key not in processed_keys: @@ -887,12 +888,16 @@ def request_params_to_args( return {first_field.name: v_}, errors_ for field in fields: - value = _get_multidict_value(field, received_params) + aliases = _get_param_aliases(field) + _, alias = next( + (pair for pair in aliases if pair[1] in received_params), aliases[0] + ) + value = _get_multidict_value(field, received_params, alias=alias) field_info = field.field_info assert isinstance(field_info, params.Param), ( "Params must be subclasses of Param" ) - loc = (field_info.in_.value, get_validation_alias(field)) + loc = (field_info.in_.value, alias) v_, errors_ = _validate_value_with_model_field( field=field, value=value, values=values, loc=loc ) @@ -1185,3 +1190,35 @@ def _get_body_field( def get_validation_alias(field: ModelField) -> str: va = getattr(field, "validation_alias", None) return va or field.alias + + +def _get_param_aliases( + field: ModelField, *, convert_underscores: bool = False +) -> list[tuple[str, str]]: + """Return (validation alias, request key) pairs in preference order.""" + alias = get_validation_alias(field) + request_alias = alias + if convert_underscores and alias == field.name: + request_alias = alias.replace("_", "-") + validation_alias = field.field_info.validation_alias + if ( + isinstance(field.field_info, params.Path) + or not isinstance(validation_alias, AliasChoices) + or any(isinstance(alias, AliasPath) for alias in validation_alias.choices) + ): + return [(alias, request_alias)] + choices = cast(list[str], validation_alias.choices) + # Explicit choices are literal names, including underscores in headers. + aliases = [(choice, choice) for choice in choices] + config = field.config or {} + if config.get("validate_by_alias") is False or config.get( + "validate_by_name", config.get("populate_by_name", False) + ): + request_name = field.name + if convert_underscores: + request_name = request_name.replace("_", "-") + if config.get("validate_by_alias") is False: + return [(field.name, request_name)] + if field.name not in choices: + aliases.append((field.name, request_name)) + return aliases diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py index 74caac44a4..aec55ae05f 100644 --- a/fastapi/openapi/utils.py +++ b/fastapi/openapi/utils.py @@ -26,8 +26,8 @@ from fastapi.dependencies.models import ( ) from fastapi.dependencies.utils import ( _get_flat_fields_from_params, + _get_param_aliases, get_flat_params, - get_validation_alias, ) from fastapi.encoders import jsonable_encoder from fastapi.exceptions import FastAPIDeprecationWarning @@ -189,24 +189,19 @@ def _get_openapi_operation_parameters( # field_info = cast(Param, field_info) if not getattr(field_info, "include_in_schema", True): continue + convert_underscores = param_type == ParamTypes.header and getattr( + field_info, "convert_underscores", default_convert_underscores + ) + alias, name = _get_param_aliases( + param, convert_underscores=convert_underscores + )[0] param_schema = get_schema_from_model_field( field=param, model_name_map=model_name_map, field_mapping=field_mapping, separate_input_output_schemas=separate_input_output_schemas, + field_alias=alias, ) - name = get_validation_alias(param) - convert_underscores = getattr( - param.field_info, - "convert_underscores", - default_convert_underscores, - ) - if ( - param_type == ParamTypes.header - and name == param.name - and convert_underscores - ): - name = param.name.replace("_", "-") parameter = { "name": name, diff --git a/tests/test_header_cookie_alias_choices.py b/tests/test_header_cookie_alias_choices.py new file mode 100644 index 0000000000..725b1764ab --- /dev/null +++ b/tests/test_header_cookie_alias_choices.py @@ -0,0 +1,649 @@ +from typing import Annotated + +import pytest +from fastapi import Cookie, FastAPI, Header, Path +from fastapi._compat import PYDANTIC_VERSION_MINOR_TUPLE +from fastapi.testclient import TestClient +from inline_snapshot import snapshot +from pydantic import AliasChoices, BaseModel, ConfigDict, Field + +app = FastAPI() + + +@app.get("/header") +def read_header( + x_value: Annotated[ + int, + Header(alias="ignored", validation_alias=AliasChoices("X-First", "X-Second")), + ], + item_values: Annotated[ + list[str], Header(validation_alias=AliasChoices("item_values", "X-Tag")) + ] = ["default"], # noqa: B006 +): + return {"value": x_value, "items": item_values} + + +class HeaderModel(BaseModel): + x_value: Annotated[ + int, + Field(alias="ignored", validation_alias=AliasChoices("X-First", "X-Second")), + ] + item_values: Annotated[ + list[str], Field(validation_alias=AliasChoices("item_values", "X-Tag")) + ] = ["default"] + + +@app.get("/header-model") +def read_header_model(data: Annotated[HeaderModel, Header()]): + return {"value": data.x_value, "items": data.item_values} + + +@app.get("/cookie") +def read_cookie( + value: Annotated[ + int, + Cookie( + alias="ignored", validation_alias=AliasChoices("session", "legacy_session") + ), + ], +): + return {"value": value} + + +class CookieModel(BaseModel): + value: Annotated[ + int, + Field( + alias="ignored", validation_alias=AliasChoices("session", "legacy_session") + ), + ] + + +@app.get("/cookie-model") +def read_cookie_model(data: Annotated[CookieModel, Cookie()]): + return {"value": data.value} + + +@app.get("/path/{value}") +def read_path( + value: Annotated[int, Path(validation_alias=AliasChoices("first", "second"))], +): + return value + + +client = TestClient(app) + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +@pytest.mark.parametrize("alias", ["x-first", "X-SECOND"]) +def test_header_choices_are_case_insensitive(path: str, alias: str): + response = client.get(path, headers={alias: "1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": ["default"]} + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +@pytest.mark.parametrize("alias", ["ITEM_VALUES", "x-tag"]) +@pytest.mark.parametrize("items", [["one"], ["one", "two"], [""]]) +def test_header_list_choices_preserve_items(path: str, alias: str, items: list[str]): + headers = [("x-first", "1")] + [(alias, item) for item in items] + response = client.get(path, headers=headers) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": items} + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +def test_repeated_scalar_header_uses_first_value(path: str): + response = client.get(path, headers=[("X-Second", "1"), ("x-second", "2")]) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": ["default"]} + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +def test_header_choices_follow_declaration_order(path: str): + response = client.get( + path, + headers=[ + ("x-second", "2"), + ("x-first", "1"), + ("x-tag", "unused"), + ("item_values", "one"), + ("item_values", "two"), + ], + ) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": ["one", "two"]} + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +@pytest.mark.parametrize( + "headers, alias, value", + [ + ({"x-first": "invalid", "x-second": "2"}, "X-First", "invalid"), + ({"x-first": "", "x-second": "2"}, "X-First", ""), + ({"x-second": "invalid"}, "X-Second", "invalid"), + ], +) +def test_invalid_header_choice_uses_selected_alias(path, headers, alias, value): + response = client.get(path, headers=headers) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "int_parsing", + "loc": ["header", alias], + "msg": "Input should be a valid integer, unable to parse string as an integer", + "input": value, + } + ] + } + + +@pytest.mark.parametrize("path", ["/header", "/header-model"]) +def test_header_missing_uses_first_choice(path: str): + response = client.get(path, headers={"ignored": "1", "x-value": "1"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["type"] == "missing" + assert response.json()["detail"][0]["loc"] == ["header", "X-First"] + + +@pytest.mark.parametrize("path", ["/cookie", "/cookie-model"]) +@pytest.mark.parametrize("alias", ["session", "legacy_session"]) +def test_cookie_choices(path: str, alias: str): + response = client.get(path, headers={"cookie": f"{alias}=1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1} + + +@pytest.mark.parametrize("path", ["/cookie", "/cookie-model"]) +def test_cookie_choices_follow_declaration_order(path: str): + response = client.get(path, headers={"cookie": "legacy_session=2; session=1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1} + + +@pytest.mark.parametrize("path", ["/cookie", "/cookie-model"]) +def test_repeated_cookie_uses_last_value(path: str): + response = client.get( + path, headers={"cookie": "legacy_session=1; legacy_session=2"} + ) + assert response.status_code == 200 + assert response.json() == {"value": 2} + + +@pytest.mark.parametrize("path", ["/cookie", "/cookie-model"]) +@pytest.mark.parametrize( + "cookie, alias, value", + [ + ("session=invalid; legacy_session=2", "session", "invalid"), + ("session=; legacy_session=2", "session", ""), + ("legacy_session=invalid", "legacy_session", "invalid"), + ], +) +def test_invalid_cookie_choice_uses_selected_alias(path, cookie, alias, value): + response = client.get(path, headers={"cookie": cookie}) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "int_parsing", + "loc": ["cookie", alias], + "msg": "Input should be a valid integer, unable to parse string as an integer", + "input": value, + } + ] + } + + +@pytest.mark.parametrize("path", ["/cookie", "/cookie-model"]) +@pytest.mark.parametrize("cookie", ["", "value=1; ignored=1", "Session=1"]) +def test_cookie_missing_and_unaccepted_names(path: str, cookie: str): + response = client.get(path, headers={"cookie": cookie}) + assert response.status_code == 422 + assert response.json()["detail"][0]["type"] == "missing" + assert response.json()["detail"][0]["loc"] == ["cookie", "session"] + + +@pytest.mark.parametrize("path", ["/", "/model"]) +def test_cookie_default(path: str): + app = FastAPI() + + class Model(BaseModel): + value: Annotated[ + int, Field(validation_alias=AliasChoices("first", "second")) + ] = 10 + + @app.get("/model") + def read_model(data: Annotated[Model, Cookie()]): + return data.value + + @app.get("/") + def read( + value: Annotated[ + int, Cookie(validation_alias=AliasChoices("first", "second")) + ] = 10, + ): + return value + + client = TestClient(app) + response = client.get(path) + assert response.status_code == 200 + assert response.json() == 10 + + +def test_header_model_does_not_duplicate_lowercase_alias_as_extra(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(extra="forbid") + host: str + value: Annotated[ + int, Field(validation_alias=AliasChoices("X-First", "X-Second")) + ] + + @app.get("/") + def read(data: Annotated[Model, Header()]): + return data + + client = TestClient(app) + client.headers.clear() + response = client.get("/", headers={"x-second": "1"}) + assert response.status_code == 200 + assert response.json() == {"host": "testserver", "value": 1} + response = client.get("/", headers={"x-first": "1", "x-second": "2"}) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["header", "x-second"], + "msg": "Extra inputs are not permitted", + "input": "2", + } + ] + } + + +def test_header_model_fields_can_share_an_unused_alias_choice(): + app = FastAPI() + + class Model(BaseModel): + names: Annotated[ + list[str], Field(validation_alias=AliasChoices("Names", "Legacy")) + ] + text: Annotated[str, Field(validation_alias=AliasChoices("Text", "Legacy"))] + + @app.get("/") + def read(data: Annotated[Model, Header()]): + return data + + client = TestClient(app) + response = client.get( + "/", headers=[("legacy", "one"), ("legacy", "two"), ("text", "hello")] + ) + assert response.status_code == 200 + assert response.json() == {"names": ["one", "two"], "text": "hello"} + + +def test_header_model_unused_alias_choice_keeps_extra_value_shape(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(extra="allow") + host: str + names: Annotated[ + list[str], Field(validation_alias=AliasChoices("Names", "Legacy")) + ] + + @app.get("/") + def read(data: Annotated[Model, Header()]): + return data + + client = TestClient(app) + client.headers.clear() + response = client.get( + "/", headers=[("names", "one"), ("names", "two"), ("legacy", "unused")] + ) + assert response.status_code == 200 + assert response.json() == { + "host": "testserver", + "names": ["one", "two"], + "legacy": "unused", + } + + +def test_cookie_model_preserves_unused_choice_as_extra(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(extra="forbid") + value: Annotated[int, Field(validation_alias=AliasChoices("first", "second"))] + + @app.get("/") + def read(data: Annotated[Model, Cookie()]): + return data + + client = TestClient(app) + response = client.get("/", headers={"cookie": "second=1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1} + response = client.get("/", headers={"cookie": "first=1; second=2"}) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["cookie", "second"], + "msg": "Extra inputs are not permitted", + "input": "2", + } + ] + } + + +@pytest.mark.parametrize( + "config, expected_value", + [ + (ConfigDict(populate_by_name=True), 2), + pytest.param( + ConfigDict(validate_by_alias=False), + 1, + marks=pytest.mark.skipif( + PYDANTIC_VERSION_MINOR_TUPLE < (2, 11), + reason="validate_by_alias requires Pydantic 2.11", + ), + ), + ], +) +@pytest.mark.parametrize( + "convert_underscores, name, rejected_name", + [(True, "x-values", "x_values"), (False, "x_values", "x-values")], +) +def test_header_model_name_fallback( + config, expected_value, convert_underscores, name, rejected_name +): + app = FastAPI() + + class Model(BaseModel): + model_config = config + x_values: Annotated[ + list[int], Field(validation_alias=AliasChoices("X-First", "X-Second")) + ] + + @app.get("/") + def read(data: Annotated[Model, Header(convert_underscores=convert_underscores)]): + return data + + client = TestClient(app) + response = client.get("/", headers={name: "1"}) + assert response.status_code == 200 + assert response.json() == {"x_values": [1]} + response = client.get("/", headers={rejected_name: "1"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["type"] == "missing" + response = client.get("/", headers={name: "1", "x-second": "2"}) + assert response.status_code == 200 + assert response.json() == {"x_values": [expected_value]} + response = client.get("/", headers={name: "invalid"}) + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["header", name, 0] + + +@pytest.mark.parametrize( + "config, expected_value", + [ + (ConfigDict(populate_by_name=True), 2), + pytest.param( + ConfigDict(validate_by_alias=False), + 1, + marks=pytest.mark.skipif( + PYDANTIC_VERSION_MINOR_TUPLE < (2, 11), + reason="validate_by_alias requires Pydantic 2.11", + ), + ), + ], +) +def test_cookie_model_name_fallback(config, expected_value): + app = FastAPI() + + class Model(BaseModel): + model_config = config + value: Annotated[int, Field(validation_alias=AliasChoices("first", "second"))] + + @app.get("/") + def read(data: Annotated[Model, Cookie()]): + return data + + client = TestClient(app) + response = client.get("/", headers={"cookie": "value=1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1} + response = client.get("/", headers={"cookie": "value=1; second=2"}) + assert response.status_code == 200 + assert response.json() == {"value": expected_value} + + +def test_path_alias_choices_are_unchanged(): + response = client.get("/path/1") + assert response.status_code == 200 + assert response.json() == 1 + + +def test_openapi(): + response = client.get("/openapi.json") + assert response.status_code == 200 + assert response.json() == snapshot( + { + "openapi": "3.1.0", + "info": {"title": "FastAPI", "version": "0.1.0"}, + "paths": { + "/header": { + "get": { + "summary": "Read Header", + "operationId": "read_header_header_get", + "parameters": [ + { + "name": "X-First", + "in": "header", + "required": True, + "schema": {"type": "integer", "title": "X-First"}, + }, + { + "name": "item_values", + "in": "header", + "required": False, + "schema": { + "type": "array", + "items": {"type": "string"}, + "default": ["default"], + "title": "Item Values", + }, + }, + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/header-model": { + "get": { + "summary": "Read Header Model", + "operationId": "read_header_model_header_model_get", + "parameters": [ + { + "name": "X-First", + "in": "header", + "required": True, + "schema": {"type": "integer", "title": "X-First"}, + }, + { + "name": "item_values", + "in": "header", + "required": False, + "schema": { + "type": "array", + "items": {"type": "string"}, + "default": ["default"], + "title": "Item Values", + }, + }, + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/cookie": { + "get": { + "summary": "Read Cookie", + "operationId": "read_cookie_cookie_get", + "parameters": [ + { + "name": "session", + "in": "cookie", + "required": True, + "schema": {"type": "integer", "title": "Session"}, + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/cookie-model": { + "get": { + "summary": "Read Cookie Model", + "operationId": "read_cookie_model_cookie_model_get", + "parameters": [ + { + "name": "session", + "in": "cookie", + "required": True, + "schema": {"type": "integer", "title": "Session"}, + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/path/{value}": { + "get": { + "summary": "Read Path", + "operationId": "read_path_path__value__get", + "parameters": [ + { + "name": "value", + "in": "path", + "required": True, + "schema": {"type": "integer", "title": "Value"}, + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + }, + "components": { + "schemas": { + "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"}, + "input": {"title": "Input"}, + "ctx": {"type": "object", "title": "Context"}, + }, + "type": "object", + "required": ["loc", "msg", "type"], + "title": "ValidationError", + }, + } + }, + } + ) diff --git a/tests/test_query_alias_choices.py b/tests/test_query_alias_choices.py new file mode 100644 index 0000000000..34a234cab8 --- /dev/null +++ b/tests/test_query_alias_choices.py @@ -0,0 +1,466 @@ +from typing import Annotated + +import pytest +from fastapi import FastAPI, Header, Query +from fastapi._compat import PYDANTIC_VERSION_MINOR_TUPLE +from fastapi.testclient import TestClient +from inline_snapshot import snapshot +from pydantic import AliasChoices, AliasPath, BaseModel, ConfigDict, Field + +app = FastAPI() + + +@app.get("/individual") +def read_individual( + value: Annotated[ + int, Query(alias="ignored", validation_alias=AliasChoices("first", "second")) + ], + items: Annotated[ + list[str], + Query(validation_alias=AliasChoices("item", "tag"), title="Chosen items"), + ] = ["default"], # noqa: B006 +): + return {"value": value, "items": items} + + +class QueryModel(BaseModel): + value: Annotated[ + int, Field(alias="ignored", validation_alias=AliasChoices("first", "second")) + ] + items: Annotated[ + list[str], + Field(validation_alias=AliasChoices("item", "tag"), title="Chosen items"), + ] = ["default"] + + +@app.get("/model") +def read_model(data: Annotated[QueryModel, Query()]): + return {"value": data.value, "items": data.items} + + +# Reusing a model across sources must preserve its aliases and schema. +@app.get("/headers") +def read_headers(data: Annotated[QueryModel, Header()]): + return data + + +client = TestClient(app) + + +@pytest.mark.parametrize("path", ["/individual", "/model"]) +@pytest.mark.parametrize("alias", ["first", "second"]) +def test_scalar_choices_and_list_default(path: str, alias: str): + response = client.get(path, params={alias: "1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": ["default"]} + + +@pytest.mark.parametrize("path", ["/individual", "/model"]) +@pytest.mark.parametrize("alias", ["item", "tag"]) +@pytest.mark.parametrize("items", [["one"], ["one", "two"], [""]]) +def test_list_choices_preserve_items(path: str, alias: str, items: list[str]): + response = client.get(path, params={"first": "1", alias: items}) + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": items} + + +@pytest.mark.parametrize("path", ["/individual", "/model"]) +def test_repeated_scalar_uses_last_value(path: str): + response = client.get(f"{path}?second=1&second=2") + assert response.status_code == 200 + assert response.json() == {"value": 2, "items": ["default"]} + + +@pytest.mark.parametrize("path", ["/individual", "/model"]) +def test_choices_follow_declaration_order(path: str): + response = client.get(f"{path}?second=2&first=1&tag=ignored&item=one&item=two") + assert response.status_code == 200 + assert response.json() == {"value": 1, "items": ["one", "two"]} + + +@pytest.mark.parametrize("path", ["/individual", "/model"]) +@pytest.mark.parametrize( + "params, alias, value", + [ + ({"first": "invalid", "second": "2"}, "first", "invalid"), + ({"first": "", "second": "2"}, "first", ""), + ({"second": "invalid"}, "second", "invalid"), + ], +) +def test_invalid_choice_uses_selected_alias(path, params, alias, value): + response = client.get(path, params=params) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "int_parsing", + "loc": ["query", alias], + "msg": "Input should be a valid integer, unable to parse string as an integer", + "input": value, + } + ] + } + + +@pytest.mark.parametrize("params", [{}, {"value": "1"}, {"ignored": "1"}]) +def test_individual_missing_uses_first_alias(params: dict[str, str]): + response = client.get("/individual", params=params) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "missing", + "loc": ["query", "first"], + "msg": "Field required", + "input": None, + } + ] + } + + +def test_inferred_query_keeps_field_alias_choices(): + app = FastAPI() + + @app.get("/") + def read( + value: Annotated[int, Field(validation_alias=AliasChoices("first", "second"))], + ): + return value + + client = TestClient(app) + response = client.get("/", params={"second": "1"}) + assert response.status_code == 200 + assert response.json() == 1 + + +def test_model_preserves_unused_choices_for_extra_validation(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(extra="forbid") + value: Annotated[int, Field(validation_alias=AliasChoices("first", "second"))] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/", params={"second": "1"}) + assert response.status_code == 200 + assert response.json() == {"value": 1} + response = client.get("/", params={"first": "1", "second": "2"}) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["query", "second"], + "msg": "Extra inputs are not permitted", + "input": "2", + } + ] + } + + +def test_model_fields_can_share_an_unused_alias_choice(): + app = FastAPI() + + class Model(BaseModel): + names: Annotated[ + list[str], Field(validation_alias=AliasChoices("names", "legacy")) + ] + text: Annotated[str, Field(validation_alias=AliasChoices("text", "legacy"))] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/", params={"legacy": ["one", "two"], "text": "hello"}) + assert response.status_code == 200 + assert response.json() == {"names": ["one", "two"], "text": "hello"} + + +def test_model_unused_alias_choice_keeps_extra_value_shape(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(extra="allow") + names: Annotated[ + list[str], Field(validation_alias=AliasChoices("names", "legacy")) + ] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/", params={"names": ["one", "two"], "legacy": "unused"}) + assert response.status_code == 200 + assert response.json() == {"names": ["one", "two"], "legacy": "unused"} + + +@pytest.mark.parametrize( + "validation_alias", [AliasChoices("item", "tag"), AliasChoices("item", "items")] +) +def test_model_list_populated_by_name(validation_alias: AliasChoices): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(populate_by_name=True) + items: Annotated[list[str], Field(validation_alias=validation_alias)] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/", params={"items": "one"}) + assert response.status_code == 200 + assert response.json() == {"items": ["one"]} + response = client.get("/", params={"items": "one", "item": "two"}) + assert response.status_code == 200 + assert response.json() == {"items": ["two"]} + + +@pytest.mark.skipif( + PYDANTIC_VERSION_MINOR_TUPLE < (2, 11), + reason="validate_by_name and validate_by_alias require Pydantic 2.11", +) +def test_model_can_disable_validation_by_alias(): + app = FastAPI() + + class Model(BaseModel): + model_config = ConfigDict(validate_by_alias=False, extra="forbid") + items: Annotated[list[str], Field(validation_alias=AliasChoices("item", "tag"))] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/", params={"items": "one"}) + assert response.status_code == 200 + assert response.json() == {"items": ["one"]} + response = client.get("/", params={"items": "one", "tag": "two"}) + assert response.status_code == 422 + assert response.json() == { + "detail": [ + { + "type": "extra_forbidden", + "loc": ["query", "tag"], + "msg": "Extra inputs are not permitted", + "input": "two", + } + ] + } + + +def test_empty_string_alias(): + app = FastAPI() + + @app.get("/") + def read(value: Annotated[int, Query(validation_alias=AliasChoices("", "other"))]): + return value + + client = TestClient(app) + response = client.get("/?=1&other=2") + assert response.status_code == 200 + assert response.json() == 1 + + +@pytest.mark.parametrize( + "validation_alias", + [AliasPath("items", 0), AliasChoices("first", AliasPath("items", 0))], +) +def test_model_alias_path_is_unchanged(validation_alias: AliasPath | AliasChoices): + app = FastAPI() + + class Model(BaseModel): + value: Annotated[int, Field(validation_alias=validation_alias)] + + @app.get("/") + def read(data: Annotated[Model, Query()]): + return data + + client = TestClient(app) + response = client.get("/?items=1&items=2") + assert response.status_code == 200 + assert response.json() == {"value": 1} + + +def test_reused_model_still_accepts_headers(): + response = client.get("/headers", headers={"second": "1"}) + assert response.status_code == 200 + assert response.json() == {"ignored": 1, "items": ["default"]} + + +def test_openapi(): + response = client.get("/openapi.json") + assert response.status_code == 200 + assert response.json() == snapshot( + { + "openapi": "3.1.0", + "info": {"title": "FastAPI", "version": "0.1.0"}, + "paths": { + "/individual": { + "get": { + "summary": "Read Individual", + "operationId": "read_individual_individual_get", + "parameters": [ + { + "name": "first", + "in": "query", + "required": True, + "schema": {"type": "integer", "title": "First"}, + }, + { + "name": "item", + "in": "query", + "required": False, + "schema": { + "type": "array", + "items": {"type": "string"}, + "title": "Chosen items", + "default": ["default"], + }, + }, + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/model": { + "get": { + "summary": "Read Model", + "operationId": "read_model_model_get", + "parameters": [ + { + "name": "first", + "in": "query", + "required": True, + "schema": {"type": "integer", "title": "First"}, + }, + { + "name": "item", + "in": "query", + "required": False, + "schema": { + "type": "array", + "items": {"type": "string"}, + "title": "Chosen items", + "default": ["default"], + }, + }, + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + "/headers": { + "get": { + "summary": "Read Headers", + "operationId": "read_headers_headers_get", + "parameters": [ + { + "name": "first", + "in": "header", + "required": True, + "schema": {"type": "integer", "title": "First"}, + }, + { + "name": "item", + "in": "header", + "required": False, + "schema": { + "type": "array", + "items": {"type": "string"}, + "title": "Chosen items", + "default": ["default"], + }, + }, + ], + "responses": { + "200": { + "description": "Successful Response", + "content": {"application/json": {"schema": {}}}, + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + }, + }, + } + }, + }, + "components": { + "schemas": { + "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"}, + "input": {"title": "Input"}, + "ctx": {"type": "object", "title": "Context"}, + }, + "type": "object", + "required": ["loc", "msg", "type"], + "title": "ValidationError", + }, + } + }, + } + )