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 BaseModel
+from pydantic import BaseModel, Json
from pydantic.fields import FieldInfo
from starlette.background import BackgroundTasks as StarletteBackgroundTasks
from starlette.concurrency import run_in_threadpool
return v_, []
+def _is_json_field(field: ModelField) -> bool:
+ return any(type(item) is Json for item in field.field_info.metadata)
+
+
def _get_multidict_value(
field: ModelField, values: Mapping[str, Any], alias: Union[str, None] = None
) -> Any:
alias = alias or get_validation_alias(field)
- if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)):
+ if (
+ (not _is_json_field(field))
+ and is_sequence_field(field)
+ and isinstance(values, (ImmutableMultiDict, Headers))
+ ):
value = values.getlist(alias)
else:
value = values.get(alias, None)
--- /dev/null
+import json
+from typing import Annotated
+
+from fastapi import Cookie, FastAPI, Form, Header, Query
+from fastapi.testclient import TestClient
+from pydantic import Json
+
+app = FastAPI()
+
+
+@app.post("/form-json-list")
+def form_json_list(items: Annotated[Json[list[str]], Form()]) -> list[str]:
+ return items
+
+
+@app.get("/query-json-list")
+def query_json_list(items: Annotated[Json[list[str]], Query()]) -> list[str]:
+ return items
+
+
+@app.get("/header-json-list")
+def header_json_list(x_items: Annotated[Json[list[str]], Header()]) -> list[str]:
+ return x_items
+
+
+@app.get("/cookie-json-list")
+def cookie_json_list(items: Annotated[Json[list[str]], Cookie()]) -> list[str]:
+ return items
+
+
+client = TestClient(app)
+
+
+def test_form_json_list():
+ response = client.post(
+ "/form-json-list", data={"items": json.dumps(["abc", "def"])}
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == ["abc", "def"]
+
+
+def test_query_json_list():
+ response = client.get(
+ "/query-json-list", params={"items": json.dumps(["abc", "def"])}
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == ["abc", "def"]
+
+
+def test_header_json_list():
+ response = client.get(
+ "/header-json-list", headers={"x-items": json.dumps(["abc", "def"])}
+ )
+ assert response.status_code == 200, response.text
+ assert response.json() == ["abc", "def"]
+
+
+def test_cookie_json_list():
+ client.cookies.set("items", json.dumps(["abc", "def"]))
+ response = client.get("/cookie-json-list")
+ assert response.status_code == 200, response.text
+ assert response.json() == ["abc", "def"]
+ client.cookies.clear()