This commit is contained in:
Javier Sánchez Castro 2026-02-13 15:28:47 +00:00 committed by GitHub
commit fdce542635
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 319 additions and 2 deletions

View File

@ -441,7 +441,7 @@ def analyze_param(
)
field_info = value
if isinstance(field_info, FieldInfo):
field_info.annotation = type_annotation
field_info.annotation = use_annotation
# Get Depends from type annotation
if depends is not None and depends.dependency is None:

View File

@ -0,0 +1,184 @@
"""Test Pydantic validators in Annotated work correctly with FastAPI.
This test ensures that Pydantic v2 validators like AfterValidator, BeforeValidator
work correctly in Annotated types with FastAPI parameters.
Context: PR #13314 broke AfterValidator support and was reverted.
We need to ensure new changes preserve validator functionality.
"""
from typing import Annotated
from fastapi import Body, FastAPI, Query
from fastapi.testclient import TestClient
from pydantic import AfterValidator, BeforeValidator
def validate_positive(v: int) -> int:
"""Validator that ensures value is positive."""
if v <= 0:
raise ValueError("must be positive")
return v
def double_value(v) -> int:
"""Validator that doubles the input value."""
# BeforeValidator receives the raw value before Pydantic coercion
# For query params, this is a string
v_int = int(v) if isinstance(v, str) else v
return v_int * 2
def strip_whitespace(v: str) -> str:
"""Validator that strips whitespace."""
return v.strip()
app = FastAPI()
@app.get("/query-after-validator")
def query_after_validator(
value: Annotated[int, AfterValidator(validate_positive), Query()],
) -> int:
return value
@app.get("/query-before-validator")
def query_before_validator(
value: Annotated[int, BeforeValidator(double_value), Query()],
) -> int:
return value
@app.post("/body-after-validator")
def body_after_validator(
value: Annotated[int, AfterValidator(validate_positive), Body()],
) -> int:
return value
@app.post("/body-before-validator")
def body_before_validator(
name: Annotated[str, BeforeValidator(strip_whitespace), Body()],
) -> str:
return name
@app.get("/query-multiple-validators")
def query_multiple_validators(
value: Annotated[
int,
BeforeValidator(double_value),
AfterValidator(validate_positive),
Query(),
],
) -> int:
return value
client = TestClient(app)
def test_query_after_validator_valid():
"""Test AfterValidator in Query with valid value."""
response = client.get("/query-after-validator?value=5")
assert response.status_code == 200, response.text
assert response.json() == 5
def test_query_after_validator_invalid():
"""Test AfterValidator in Query with invalid value."""
response = client.get("/query-after-validator?value=-1")
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
{
"type": "value_error",
"loc": ["query", "value"],
"msg": "Value error, must be positive",
"input": "-1",
"ctx": {"error": {}},
}
]
}
def test_query_before_validator():
"""Test BeforeValidator in Query doubles the value."""
response = client.get("/query-before-validator?value=5")
assert response.status_code == 200, response.text
assert response.json() == 10 # 5 * 2
def test_body_after_validator_valid():
"""Test AfterValidator in Body with valid value."""
# Body() without embed means send the value directly, not wrapped in an object
response = client.post("/body-after-validator", json=10)
assert response.status_code == 200, response.text
assert response.json() == 10
def test_body_after_validator_invalid():
"""Test AfterValidator in Body with invalid value."""
response = client.post("/body-after-validator", json=-5)
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
{
"type": "value_error",
"loc": ["body"],
"msg": "Value error, must be positive",
"input": -5,
"ctx": {"error": {}},
}
]
}
def test_body_before_validator():
"""Test BeforeValidator in Body strips whitespace."""
response = client.post("/body-before-validator", json=" hello ")
assert response.status_code == 200, response.text
assert response.json() == "hello"
def test_query_multiple_validators():
"""Test multiple validators work in correct order."""
# Input: 3 → BeforeValidator doubles to 6 → AfterValidator checks positive
response = client.get("/query-multiple-validators?value=3")
assert response.status_code == 200, response.text
assert response.json() == 6
# Input: -1 → BeforeValidator doubles to -2 → AfterValidator fails
response = client.get("/query-multiple-validators?value=-1")
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
{
"type": "value_error",
"loc": ["query", "value"],
"msg": "Value error, must be positive",
"input": "-1",
"ctx": {"error": {}},
}
]
}
def test_query_multiple_validators_zero():
"""Test multiple validators with zero (edge case)."""
# Input: 0 → BeforeValidator doubles to 0 → AfterValidator fails (not positive)
response = client.get("/query-multiple-validators?value=0")
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
{
"type": "value_error",
"loc": ["query", "value"],
"msg": "Value error, must be positive",
"input": "0",
"ctx": {"error": {}},
}
]
}

View File

@ -0,0 +1,133 @@
import json
from typing import Annotated, Any
from fastapi import FastAPI, Form
from fastapi.testclient import TestClient
from pydantic import Json
app = FastAPI()
@app.post("/form-json-equals-syntax")
def form_json_equals_syntax(data: Json[dict[str, Any]] = Form()) -> dict[str, Any]:
return data
@app.post("/form-json-annotated-syntax")
def form_json_annotated_syntax(
data: Annotated[Json[dict[str, Any]], Form()],
) -> dict[str, Any]:
return data
@app.post("/form-json-list-equals")
def form_json_list_equals(items: Json[list[str]] = Form()) -> list[str]:
return items
@app.post("/form-json-list-annotated")
def form_json_list_annotated(
items: Annotated[Json[list[str]], Form()],
) -> list[str]:
return items
@app.post("/form-json-nested-equals")
def form_json_nested_equals(
config: Json[dict[str, list[int]]] = Form(),
) -> dict[str, list[int]]:
return config
@app.post("/form-json-nested-annotated")
def form_json_nested_annotated(
config: Annotated[Json[dict[str, list[int]]], Form()],
) -> dict[str, list[int]]:
return config
client = TestClient(app)
def test_form_json_equals_syntax():
test_data = {"key": "value", "nested": {"inner": 123}}
response = client.post(
"/form-json-equals-syntax", data={"data": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_annotated_syntax():
test_data = {"key": "value", "nested": {"inner": 123}}
response = client.post(
"/form-json-annotated-syntax", data={"data": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_list_equals():
test_data = ["abc", "def", "ghi"]
response = client.post(
"/form-json-list-equals", data={"items": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_list_annotated():
test_data = ["abc", "def", "ghi"]
response = client.post(
"/form-json-list-annotated", data={"items": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_nested_equals():
test_data = {"groups": [1, 2, 3], "ids": [4, 5]}
response = client.post(
"/form-json-nested-equals", data={"config": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_nested_annotated():
test_data = {"groups": [1, 2, 3], "ids": [4, 5]}
response = client.post(
"/form-json-nested-annotated", data={"config": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_invalid_json_equals():
response = client.post("/form-json-equals-syntax", data={"data": "not valid json{"})
assert response.status_code == 422
def test_form_json_invalid_json_annotated():
response = client.post(
"/form-json-annotated-syntax", data={"data": "not valid json{"}
)
assert response.status_code == 422
def test_form_json_empty_dict_equals():
test_data = {}
response = client.post(
"/form-json-equals-syntax", data={"data": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data
def test_form_json_empty_list_equals():
test_data = []
response = client.post(
"/form-json-list-equals", data={"items": json.dumps(test_data)}
)
assert response.status_code == 200, response.text
assert response.json() == test_data

View File

@ -73,7 +73,7 @@ def test_openapi_schema(client: TestClient) -> None:
"content": {
"application/json": {
"schema": {
"anyOf": [
"oneOf": [
{"$ref": "#/components/schemas/Cat"},
{"$ref": "#/components/schemas/Dog"},
],