Browse Source

fix: honor AliasChoices for direct request params

pull/15341/head
OussemaNehdi 3 months ago
parent
commit
effb0fb192
  1. 91
      fastapi/dependencies/utils.py
  2. 127
      tests/test_request_params/test_query/test_required_str.py

91
fastapi/dependencies/utils.py

@ -60,7 +60,7 @@ from fastapi.logger import logger
from fastapi.security.oauth2 import SecurityScopes from fastapi.security.oauth2 import SecurityScopes
from fastapi.types import DependencyCacheKey from fastapi.types import DependencyCacheKey
from fastapi.utils import create_model_field, get_path_param_names from fastapi.utils import create_model_field, get_path_param_names
from pydantic import BaseModel, Json from pydantic import AliasChoices, AliasPath, BaseModel, Json
from pydantic.fields import FieldInfo from pydantic.fields import FieldInfo
from starlette.background import BackgroundTasks as StarletteBackgroundTasks from starlette.background import BackgroundTasks as StarletteBackgroundTasks
from starlette.concurrency import run_in_threadpool from starlette.concurrency import run_in_threadpool
@ -750,18 +750,72 @@ def _is_json_field(field: ModelField) -> bool:
return any(type(item) is Json for item in field.field_info.metadata) return any(type(item) is Json for item in field.field_info.metadata)
def _get_alias_path_name(alias_path: AliasPath) -> str | None:
# Query/header/cookie/form/file values are flat mappings, so only single-key
# alias paths can be resolved here.
if len(alias_path.path) == 1 and isinstance(alias_path.path[0], str):
return alias_path.path[0]
return None
def get_validation_aliases(field: ModelField) -> tuple[str, ...]:
validation_alias = getattr(field.field_info, "validation_alias", None)
if validation_alias is None:
return (field.alias,)
if isinstance(validation_alias, str):
return (validation_alias,)
if isinstance(validation_alias, AliasPath):
alias = _get_alias_path_name(validation_alias)
if alias is not None:
return (alias,)
return (field.alias,)
if isinstance(validation_alias, AliasChoices):
aliases: list[str] = []
for alias_choice in validation_alias.choices:
if isinstance(alias_choice, str):
aliases.append(alias_choice)
elif isinstance(alias_choice, AliasPath):
alias = _get_alias_path_name(alias_choice)
if alias is not None:
aliases.append(alias)
if aliases:
# Keep first-seen order for predictable precedence.
return tuple(dict.fromkeys(aliases))
return (field.alias,)
def _get_multidict_value( def _get_multidict_value(
field: ModelField, values: Mapping[str, Any], alias: str | None = None field: ModelField,
values: Mapping[str, Any],
alias: str | Sequence[str] | None = None,
) -> Any: ) -> Any:
alias = alias or get_validation_alias(field) if alias is None:
if ( aliases = get_validation_aliases(field)
(not _is_json_field(field)) elif isinstance(alias, str):
and field_annotation_is_sequence(field.field_info.annotation) aliases = (alias,)
and isinstance(values, (ImmutableMultiDict, Headers))
):
value = values.getlist(alias)
else: else:
value = values.get(alias, None) aliases = tuple(alias)
value: Any = None
for alias_name in aliases:
if (
(not _is_json_field(field))
and field_annotation_is_sequence(field.field_info.annotation)
and isinstance(values, (ImmutableMultiDict, Headers))
):
candidate = values.getlist(alias_name)
if candidate:
value = candidate
break
else:
if alias_name in values:
value = values.get(alias_name)
break
if ( if (
value is None value is None
or ( or (
@ -822,10 +876,11 @@ def request_params_to_args(
alias = get_validation_alias(field) alias = get_validation_alias(field)
if alias == field.name: if alias == field.name:
alias = alias.replace("_", "-") alias = alias.replace("_", "-")
value = _get_multidict_value(field, received_params, alias=alias) aliases_to_process = (alias,) if alias is not None else get_validation_aliases(field)
value = _get_multidict_value(field, received_params, alias=aliases_to_process)
if value is not None: if value is not None:
params_to_process[get_validation_alias(field)] = value params_to_process[get_validation_alias(field)] = value
processed_keys.add(alias or get_validation_alias(field)) processed_keys.update(aliases_to_process)
for key in received_params.keys(): for key in received_params.keys():
if key not in processed_keys: if key not in processed_keys:
@ -937,7 +992,9 @@ async def _extract_form_body(
value = serialize_sequence_value(field=field, value=results) value = serialize_sequence_value(field=field, value=results)
if value is not None: if value is not None:
values[get_validation_alias(field)] = value values[get_validation_alias(field)] = value
field_aliases = {get_validation_alias(field) for field in body_fields} field_aliases = {
alias for field in body_fields for alias in get_validation_aliases(field)
}
for key in received_body.keys(): for key in received_body.keys():
if key not in field_aliases: if key not in field_aliases:
param_values = received_body.getlist(key) param_values = received_body.getlist(key)
@ -983,7 +1040,10 @@ async def request_body_to_args(
value: Any | None = None value: Any | None = None
if body_to_process is not None and not isinstance(body_to_process, bytes): if body_to_process is not None and not isinstance(body_to_process, bytes):
try: try:
value = body_to_process.get(get_validation_alias(field)) for alias in get_validation_aliases(field):
if alias in body_to_process:
value = body_to_process.get(alias)
break
# If the received body is a list, not a dict # If the received body is a list, not a dict
except AttributeError: except AttributeError:
errors.append(get_missing_field_error(loc)) errors.append(get_missing_field_error(loc))
@ -1053,5 +1113,4 @@ def get_body_field(
def get_validation_alias(field: ModelField) -> str: def get_validation_alias(field: ModelField) -> str:
va = getattr(field, "validation_alias", None) return get_validation_aliases(field)[0]
return va or field.alias

127
tests/test_request_params/test_query/test_required_str.py

@ -5,7 +5,7 @@ from dirty_equals import IsOneOf
from fastapi import FastAPI, Query from fastapi import FastAPI, Query
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from inline_snapshot import snapshot from inline_snapshot import snapshot
from pydantic import BaseModel, Field from pydantic import AliasChoices, BaseModel, Field
app = FastAPI() app = FastAPI()
@ -274,6 +274,131 @@ def test_required_validation_alias_by_validation_alias(path: str):
assert response.json() == {"p": "hello"} assert response.json() == {"p": "hello"}
# =====================================================================================
# Validation alias choices
@app.get("/required-validation-alias-choices")
def read_required_validation_alias_choices(
p: Annotated[
str,
Query(validation_alias=AliasChoices("p_val_alias", "p_fallback_alias")),
],
):
return {"p": p}
class QueryModelRequiredValidationAliasChoices(BaseModel):
p: str = Field(validation_alias=AliasChoices("p_val_alias", "p_fallback_alias"))
@app.get("/model-required-validation-alias-choices")
def read_model_required_validation_alias_choices(
p: Annotated[QueryModelRequiredValidationAliasChoices, Query()],
):
return {"p": p.p}
@pytest.mark.parametrize(
"path",
[
"/required-validation-alias-choices",
"/model-required-validation-alias-choices",
],
)
def test_required_validation_alias_choices_schema(path: str):
assert app.openapi()["paths"][path]["get"]["parameters"] == snapshot(
[
{
"required": True,
"schema": {"title": "P", "type": "string"},
"name": "p_val_alias",
"in": "query",
}
]
)
@pytest.mark.parametrize(
"path",
[
"/required-validation-alias-choices",
"/model-required-validation-alias-choices",
],
)
def test_required_validation_alias_choices_missing(path: str):
client = TestClient(app)
response = client.get(path)
assert response.status_code == 422
assert response.json() == {
"detail": [
{
"type": "missing",
"loc": [
"query",
"p_val_alias",
],
"msg": "Field required",
"input": IsOneOf(None, {}),
}
]
}
@pytest.mark.parametrize(
"path",
[
"/required-validation-alias-choices",
"/model-required-validation-alias-choices",
],
)
def test_required_validation_alias_choices_by_name(path: str):
client = TestClient(app)
response = client.get(f"{path}?p=hello")
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
{
"type": "missing",
"loc": ["query", "p_val_alias"],
"msg": "Field required",
"input": IsOneOf(None, {"p": "hello"}),
}
]
}
@pytest.mark.parametrize(
"path",
[
"/required-validation-alias-choices",
"/model-required-validation-alias-choices",
],
)
def test_required_validation_alias_choices_by_primary_alias(path: str):
client = TestClient(app)
response = client.get(f"{path}?p_val_alias=hello")
assert response.status_code == 200, response.text
assert response.json() == {"p": "hello"}
@pytest.mark.parametrize(
"path",
[
"/required-validation-alias-choices",
"/model-required-validation-alias-choices",
],
)
def test_required_validation_alias_choices_by_fallback_alias(path: str):
client = TestClient(app)
response = client.get(f"{path}?p_fallback_alias=hello")
assert response.status_code == 200, response.text
assert response.json() == {"p": "hello"}
# ===================================================================================== # =====================================================================================
# Alias and validation alias # Alias and validation alias

Loading…
Cancel
Save