Nested Pydantic Models for Get parameters using Query in version 0.115.0

[seabean1999]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Annotated
from fastapi import Query, APIRouter, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class ReqData(BaseModel):
	name: str = Field(...)

class ReqProjectList(BaseModel):
	search: str | None = Field(None, max_length=50)
	limit: int = Field(20, ge=1, le=100)
	offset: int = Field(0, ge=0)
	filter: ReqData | None = Field(None)

@app.get("/all", description="xxxx")
async def get_all_projects(req: Annotated[ReqProjectList, Query()]):
    return {"Hello": "World"}

Description

Hi，it seems the Nested Pydantic Models don't work for parameters using Query in the latest version. Is this the expected design?
When the server code runs as above, a request for the router curl https://{ip}:{port}/all?search=com&limit=20&offset=0&filter=%7B%22name%22:%22xxx%22%7D} will recieve a validation exception:

[{'type': 'model_attributes_type', 'loc': ('query', 'filter'), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': '{"name":"xxx"}'}]

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.115.0

Pydantic Version

2.5.3

Python Version

3.12

Additional Context

No response

[ceb10n]

apparently this is not possible at the moment 😢

When FastAPI tries to get values from the body, it will check if it's a json, and will get a valid dict from it:

if body_field:
    ...
    body_bytes = await request.body()
    if body_bytes:
        json_body: Any = Undefined
        content_type_value = request.headers.get("content-type")
        ...
        else:
            message = email.message.Message()
            message["content-type"] = content_type_value
            if message.get_content_maintype() == "application":
                subtype = message.get_content_subtype()
                if subtype == "json" or subtype.endswith("+json"):
                    json_body = await request.json()
        if json_body != Undefined:
            body = json_body
        else:
            body = body_bytes

And this will give a valid json/dict to FastAPI try to validate it with Pydantic's TypeAdapter.

But with Query, it will get the values from Starlette's QueryParams:

QueryParams('search=test&limit=10&offset=1&filter=%7B%22name%22%3A%22teste%22%7D')

And this will return a string with value '{"name":"teste"}' for the filter field.

The ModelField validate function calls TypeAdapter.validate_python function with the value:

{'search': 'test', 'limit': '10', 'offset': '1', 'filter': '{"name":"teste"}'}

see that filter is a string, not a dict.

It's because FastAPI is not looking for a valid json here, it will simply use what Starlette's QueryParams is giving to it.

You can check it in request_params_to_args function, and then in _get_multidict_value function that lives inside dependencies/utils.py

I don't know if it's already addressed by @tiangolo, but maybe _get_mutidict_value can add some validation to check if the type from the ModelField is and object, and try to process it as a json, not a string.

maybe it's the case to open an issue?

[sinisaos]

I don't know if this is a satisfactory solution, but with changes in request_params_to_args it is possible to use nested models (deeper than one level) in query params. Something like this.

Example code

def request_params_to_args(
    fields: Sequence[ModelField],
    received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[Any]]:
    values: Dict[str, Any] = {}
    errors: List[Dict[str, Any]] = []

    if not fields:
        return values, errors

    first_field = fields[0]
    fields_to_extract = fields
    single_not_embedded_field = False
    if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):
        fields_to_extract = get_cached_model_fields(first_field.type_)
        single_not_embedded_field = True

    params_to_process: Dict[str, Any] = {}

    processed_keys = set()

    for field in fields_to_extract:
        alias = None
        try:
            value = _get_multidict_value(field, received_params, alias=alias)
            # check if field is a subclass of BaseModel and handle nested Pydantic
            # model (deeper than one level) in query params
            if (
                lenient_issubclass(get_args(field.field_info.annotation)[0], BaseModel)
                and value is not None
            ):
                # parse a valid JSON string and convert it into a dict
                params_to_process[field.alias] = json.loads(value) # use alias instead of name PR 12411
            else:
                params_to_process[field.alias] = value # use alias instead of name PR 12411
        except IndexError:
            pass
        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", True)
            if convert_underscores:
                alias = (
                    field.alias
                    if field.alias != field.name
                    else field.name.replace("_", "-")
                )
        processed_keys.add(alias or field.alias)
        processed_keys.add(field.name)

    for key, value in received_params.items():
        if key not in processed_keys:
            params_to_process[key] = value

    if single_not_embedded_field:
        field_info = first_field.field_info
        assert isinstance(
            field_info, params.Param
        ), "Params must be subclasses of Param"
        loc: Tuple[str, ...] = (field_info.in_.value,)
        v_, errors_ = _validate_value_with_model_field(
            field=first_field, value=params_to_process, values=values, loc=loc
        )
        return {first_field.name: v_}, errors_

    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)
        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

I see at least one problem with this solution. If we have large models, by parsing the json from the query string, we can exceed the characters of the browser query parameter (2048). Sorry if I missed your point.

[YuriiMotov]

Duplicate of #12214