From d7945291113d504030cdfb3762a960574b73e7fc Mon Sep 17 00:00:00 2001 From: Dan Johns Date: Tue, 24 Mar 2026 10:52:24 +0000 Subject: [PATCH 1/2] handle Union and RootModel sequences in Body validation --- .../middlewares/openapi_validation.py | 37 ++++- .../test_openapi_validation_middleware.py | 148 +++++++++++++++++- 2 files changed, 182 insertions(+), 3 deletions(-) diff --git a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py index 0a0fc3c8075..c80984b6b3c 100644 --- a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py +++ b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py @@ -3,10 +3,11 @@ import dataclasses import json import logging -from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence, cast +from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence, Union, cast from urllib.parse import parse_qs from pydantic import BaseModel +from typing_extensions import get_args, get_origin from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler from aws_lambda_powertools.event_handler.openapi.compat import ( @@ -25,6 +26,7 @@ ResponseValidationError, ) from aws_lambda_powertools.event_handler.openapi.params import Param +from aws_lambda_powertools.event_handler.openapi.types import UnionType if TYPE_CHECKING: from pydantic.fields import FieldInfo @@ -431,9 +433,40 @@ def _handle_missing_field_value( values[field.name] = field.get_default() +def _is_or_contains_sequence(annotation: Any) -> bool: + """ + Check if annotation is a sequence or Union/RootModel containing a sequence. + + This function handles complex type annotations like: + - Union[Model, List[Model]] - checks if any Union member is a sequence + - RootModel[List[Model]] - checks if the RootModel wraps a sequence + """ + # Direct sequence check + if field_annotation_is_sequence(annotation): + return True + + # Check Union members for any sequence types + origin = get_origin(annotation) + if origin is Union or origin is UnionType: + for arg in get_args(annotation): + if field_annotation_is_sequence(arg): + return True + + # Check if it's a RootModel wrapping a sequence + if lenient_issubclass(annotation, BaseModel): + # Check if it's a RootModel by looking for __pydantic_root_model__ + if getattr(annotation, "__pydantic_root_model__", False): + # Get the inner type from model_fields['root'] + if hasattr(annotation, "model_fields") and "root" in annotation.model_fields: + root_field = annotation.model_fields["root"] + return field_annotation_is_sequence(root_field.annotation) + + return False + + def _normalize_field_value(value: Any, field_info: FieldInfo) -> Any: """Normalize field value, converting lists to single values for non-sequence fields.""" - if field_annotation_is_sequence(field_info.annotation): + if _is_or_contains_sequence(field_info.annotation): return value elif isinstance(value, list) and value: return value[0] diff --git a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py index 93baef283ba..b99a71bc70f 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py @@ -7,7 +7,7 @@ from typing import Any, Dict, List, Literal, Optional, Tuple, Union import pytest -from pydantic import AfterValidator, Base64UrlStr, BaseModel, ConfigDict, Field, StringConstraints, alias_generators +from pydantic import AfterValidator, Base64UrlStr, BaseModel, ConfigDict, Field, RootModel, StringConstraints, alias_generators from typing_extensions import Annotated from aws_lambda_powertools.event_handler import ( @@ -2833,3 +2833,149 @@ def handler(query_dt: datetime.datetime): # THEN validation should fail because the encoded string is not a valid datetime result = app(raw_event, {}) assert result["statusCode"] == 422 + + +def test_validate_union_single_or_list_body_with_list(gw_event): + """Test that Union[Model, List[Model]] correctly handles a list of items""" + # GIVEN an APIGatewayRestResolver with validation enabled + app = APIGatewayRestResolver(enable_validation=True) + + class Item(BaseModel): + name: str + value: int + + # WHEN a handler is defined with Union[Model, List[Model]] body parameter + @app.post("/items") + def handler(items: Annotated[Union[Item, List[Item]], Body()]) -> Dict[str, Any]: + # Should receive the full list, not just the first element + if isinstance(items, list): + return {"count": len(items), "items": [item.model_dump() for item in items]} + else: + return {"count": 1, "items": [items.model_dump()]} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/items" + # Send a list of items + gw_event["body"] = json.dumps([ + {"name": "item1", "value": 10}, + {"name": "item2", "value": 20}, + {"name": "item3", "value": 30}, + ]) + + # THEN the handler should receive all items in the list, not just the first one + result = app(gw_event, {}) + assert result["statusCode"] == 200 + body = json.loads(result["body"]) + assert body["count"] == 3 + assert len(body["items"]) == 3 + assert body["items"][0]["name"] == "item1" + assert body["items"][1]["name"] == "item2" + assert body["items"][2]["name"] == "item3" + + +def test_validate_union_single_or_list_body_with_single(gw_event): + """Test that Union[Model, List[Model]] correctly handles a single item""" + # GIVEN an APIGatewayRestResolver with validation enabled + app = APIGatewayRestResolver(enable_validation=True) + + class Item(BaseModel): + name: str + value: int + + # WHEN a handler is defined with Union[Model, List[Model]] body parameter + @app.post("/items") + def handler(items: Annotated[Union[Item, List[Item]], Body()]) -> Dict[str, Any]: + if isinstance(items, list): + return {"count": len(items), "items": [item.model_dump() for item in items]} + else: + return {"count": 1, "items": [items.model_dump()]} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/items" + # Send a single item + gw_event["body"] = json.dumps({"name": "single_item", "value": 42}) + + # THEN the handler should receive the single item + result = app(gw_event, {}) + assert result["statusCode"] == 200 + body = json.loads(result["body"]) + assert body["count"] == 1 + assert len(body["items"]) == 1 + assert body["items"][0]["name"] == "single_item" + assert body["items"][0]["value"] == 42 + + +def test_validate_rootmodel_list_body(gw_event): + """Test that RootModel[List[Model]] correctly handles a list of items""" + # GIVEN an APIGatewayRestResolver with validation enabled + app = APIGatewayRestResolver(enable_validation=True) + + class Item(BaseModel): + name: str + value: int + + class ItemCollection(RootModel[List[Item]]): + root: List[Item] + + # WHEN a handler is defined with RootModel[List[Model]] body parameter + @app.post("/items") + def handler(collection: Annotated[ItemCollection, Body()]) -> Dict[str, Any]: + # collection.root should contain the full list + items = collection.root + return {"count": len(items), "items": [item.model_dump() for item in items]} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/items" + # Send a list of items + gw_event["body"] = json.dumps([ + {"name": "item1", "value": 100}, + {"name": "item2", "value": 200}, + ]) + + # THEN the handler should receive all items in the collection + result = app(gw_event, {}) + assert result["statusCode"] == 200 + body = json.loads(result["body"]) + assert body["count"] == 2 + assert len(body["items"]) == 2 + assert body["items"][0]["name"] == "item1" + assert body["items"][0]["value"] == 100 + assert body["items"][1]["name"] == "item2" + assert body["items"][1]["value"] == 200 + + +def test_validate_nested_union_with_sequence(gw_event): + """Test that nested Union types containing sequences are handled correctly""" + # GIVEN an APIGatewayRestResolver with validation enabled + app = APIGatewayRestResolver(enable_validation=True) + + class Person(BaseModel): + name: str + age: int + + # WHEN a handler is defined with a complex Union including List + @app.post("/people") + def handler( + data: Annotated[Union[str, List[Person], Person], Body()], + ) -> Dict[str, Any]: + if isinstance(data, str): + return {"type": "string", "value": data} + elif isinstance(data, list): + return {"type": "list", "count": len(data)} + else: + return {"type": "person", "name": data.name} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/people" + # Send a list + gw_event["body"] = json.dumps([ + {"name": "Alice", "age": 30}, + {"name": "Bob", "age": 25}, + ]) + + # THEN the handler should receive the full list + result = app(gw_event, {}) + assert result["statusCode"] == 200 + body = json.loads(result["body"]) + assert body["type"] == "list" + assert body["count"] == 2 From a6c81662fbcda4932160ac794adc69aecdf03fdf Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Thu, 26 Mar 2026 13:45:41 +0000 Subject: [PATCH 2/2] fix(event-handler): make sequence detection recursive for nested Union/RootModel The original fix only checked one level deep. This makes _is_or_contains_sequence recursive so it catches: - Optional[RootModel[List[Model]]] (Union containing RootModel) - RootModel[Union[Model, List[Model]]] (RootModel wrapping Union) Adds 16 regression tests covering edge cases: Optional, empty list, single-element list, pipe syntax, cross-resolver, large payloads, etc. Co-Authored-By: Claude Opus 4.6 --- .../middlewares/openapi_validation.py | 21 +- .../test_openapi_validation_middleware.py | 376 +++++++++++++++++- 2 files changed, 373 insertions(+), 24 deletions(-) diff --git a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py index c80984b6b3c..dbfc0a6f9d7 100644 --- a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py +++ b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py @@ -438,28 +438,29 @@ def _is_or_contains_sequence(annotation: Any) -> bool: Check if annotation is a sequence or Union/RootModel containing a sequence. This function handles complex type annotations like: + - List[Model] - direct sequence - Union[Model, List[Model]] - checks if any Union member is a sequence + - Optional[List[Model]] - Union[List[Model], None] - RootModel[List[Model]] - checks if the RootModel wraps a sequence + - Optional[RootModel[List[Model]]] - Union member that is a RootModel + - RootModel[Union[Model, List[Model]]] - RootModel wrapping a Union with a sequence """ # Direct sequence check if field_annotation_is_sequence(annotation): return True - # Check Union members for any sequence types + # Check Union members — recurse so we catch RootModel inside Union origin = get_origin(annotation) if origin is Union or origin is UnionType: for arg in get_args(annotation): - if field_annotation_is_sequence(arg): + if _is_or_contains_sequence(arg): return True - # Check if it's a RootModel wrapping a sequence - if lenient_issubclass(annotation, BaseModel): - # Check if it's a RootModel by looking for __pydantic_root_model__ - if getattr(annotation, "__pydantic_root_model__", False): - # Get the inner type from model_fields['root'] - if hasattr(annotation, "model_fields") and "root" in annotation.model_fields: - root_field = annotation.model_fields["root"] - return field_annotation_is_sequence(root_field.annotation) + # Check if it's a RootModel wrapping a sequence (or Union containing a sequence) + if lenient_issubclass(annotation, BaseModel) and getattr(annotation, "__pydantic_root_model__", False): + if hasattr(annotation, "model_fields") and "root" in annotation.model_fields: + root_annotation = annotation.model_fields["root"].annotation + return _is_or_contains_sequence(root_annotation) return False diff --git a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py index b99a71bc70f..21bc9b26e0a 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py @@ -7,7 +7,16 @@ from typing import Any, Dict, List, Literal, Optional, Tuple, Union import pytest -from pydantic import AfterValidator, Base64UrlStr, BaseModel, ConfigDict, Field, RootModel, StringConstraints, alias_generators +from pydantic import ( + AfterValidator, + Base64UrlStr, + BaseModel, + ConfigDict, + Field, + RootModel, + StringConstraints, + alias_generators, +) from typing_extensions import Annotated from aws_lambda_powertools.event_handler import ( @@ -2856,11 +2865,13 @@ def handler(items: Annotated[Union[Item, List[Item]], Body()]) -> Dict[str, Any] gw_event["httpMethod"] = "POST" gw_event["path"] = "/items" # Send a list of items - gw_event["body"] = json.dumps([ - {"name": "item1", "value": 10}, - {"name": "item2", "value": 20}, - {"name": "item3", "value": 30}, - ]) + gw_event["body"] = json.dumps( + [ + {"name": "item1", "value": 10}, + {"name": "item2", "value": 20}, + {"name": "item3", "value": 30}, + ], + ) # THEN the handler should receive all items in the list, not just the first one result = app(gw_event, {}) @@ -2927,10 +2938,12 @@ def handler(collection: Annotated[ItemCollection, Body()]) -> Dict[str, Any]: gw_event["httpMethod"] = "POST" gw_event["path"] = "/items" # Send a list of items - gw_event["body"] = json.dumps([ - {"name": "item1", "value": 100}, - {"name": "item2", "value": 200}, - ]) + gw_event["body"] = json.dumps( + [ + {"name": "item1", "value": 100}, + {"name": "item2", "value": 200}, + ], + ) # THEN the handler should receive all items in the collection result = app(gw_event, {}) @@ -2968,10 +2981,12 @@ def handler( gw_event["httpMethod"] = "POST" gw_event["path"] = "/people" # Send a list - gw_event["body"] = json.dumps([ - {"name": "Alice", "age": 30}, - {"name": "Bob", "age": 25}, - ]) + gw_event["body"] = json.dumps( + [ + {"name": "Alice", "age": 30}, + {"name": "Bob", "age": 25}, + ], + ) # THEN the handler should receive the full list result = app(gw_event, {}) @@ -2979,3 +2994,336 @@ def handler( body = json.loads(result["body"]) assert body["type"] == "list" assert body["count"] == 2 + + +# ──────────────────────────────────────────────────────────────────── +# Regression tests for Union / RootModel / Optional sequence body +# See: https://github.com/aws-powertools/powertools-lambda-python/issues/8057 +# ──────────────────────────────────────────────────────────────────── + + +class _Item(BaseModel): + name: str + value: int + + +class _ItemCollection(RootModel[List[_Item]]): + pass + + +_THREE_ITEMS = [ + {"name": "a", "value": 1}, + {"name": "b", "value": 2}, + {"name": "c", "value": 3}, +] + + +def _post_json(app, path, payload): + """Helper: build a minimal APIGW REST event, POST JSON, return parsed result.""" + from tests.functional.utils import load_event + + event = load_event("apiGatewayProxyEvent.json") + event["httpMethod"] = "POST" + event["path"] = path + event["body"] = json.dumps(payload) + result = app(event, {}) + return result["statusCode"], json.loads(result["body"]) + + +# ---------- Optional[List[Model]] ---------- + + +def test_optional_list_body_with_list(): + """Optional[List[Model]] must preserve the full list.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[List[_Item]], Body()]) -> Dict[str, Any]: + assert isinstance(items, list) + return {"count": len(items)} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +def test_optional_list_body_with_none(): + """Optional[List[Model]] must accept a null body gracefully.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[List[_Item]], Body()] = None) -> Dict[str, Any]: + return {"received_none": items is None} + + status, body = _post_json(app, "/items", None) + assert status == 200 + assert body["received_none"] is True + + +# ---------- Optional[Union[Model, List[Model]]] ---------- + + +def test_optional_union_model_or_list_with_list(): + """Optional[Union[Model, List[Model]]] — send list, get full list.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[Union[_Item, List[_Item]]], Body()]) -> Dict[str, Any]: + assert isinstance(items, list) + return {"count": len(items)} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +def test_optional_union_model_or_list_with_single(): + """Optional[Union[Model, List[Model]]] — send single obj, get single obj.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[Union[_Item, List[_Item]]], Body()]) -> Dict[str, Any]: + assert not isinstance(items, list) + return {"name": items.name} + + status, body = _post_json(app, "/items", {"name": "solo", "value": 99}) + assert status == 200 + assert body["name"] == "solo" + + +def test_optional_union_model_or_list_with_none(): + """Optional[Union[Model, List[Model]]] — send null, get None.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[Union[_Item, List[_Item]]], Body()] = None) -> Dict[str, Any]: + return {"is_none": items is None} + + status, body = _post_json(app, "/items", None) + assert status == 200 + assert body["is_none"] is True + + +# ---------- List[Model] directly (no Union / Optional) ---------- + + +def test_plain_list_body_preserves_all_items(): + """List[Model] — baseline: must never truncate.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[List[_Item], Body()]) -> Dict[str, Any]: + return {"count": len(items)} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +# ---------- Empty list ---------- + + +def test_union_model_or_list_with_empty_list(): + """Union[Model, List[Model]] with [] — must not crash on value[0].""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]: + if isinstance(items, list): + return {"count": len(items)} + return {"count": 1} + + status, body = _post_json(app, "/items", []) + assert status == 200 + assert body["count"] == 0 + + +def test_plain_list_with_empty_list(): + """List[Model] with [] — must accept empty list.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[List[_Item], Body()]) -> Dict[str, Any]: + return {"count": len(items)} + + status, body = _post_json(app, "/items", []) + assert status == 200 + assert body["count"] == 0 + + +# ---------- Single-element list (boundary) ---------- + + +def test_union_model_or_list_with_single_element_list(): + """Union[Model, List[Model]] with [single_item] — must NOT unwrap to scalar.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]: + if isinstance(items, list): + return {"type": "list", "count": len(items)} + return {"type": "single"} + + status, body = _post_json(app, "/items", [{"name": "only", "value": 1}]) + assert status == 200 + # Pydantic may match as single Item or list — either is valid, + # but it must NOT crash or lose data + assert body.get("count", 1) == 1 + + +# ---------- Union with primitive sequences ---------- + + +def test_union_str_or_list_dict(): + """Union[str, List[dict]] — list of dicts must arrive intact.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/data") + def handler(data: Annotated[Union[str, List[Dict[str, Any]]], Body()]) -> Dict[str, Any]: + if isinstance(data, list): + return {"type": "list", "count": len(data)} + return {"type": "str"} + + payload = [{"key": "v1"}, {"key": "v2"}] + status, body = _post_json(app, "/data", payload) + assert status == 200 + assert body["type"] == "list" + assert body["count"] == 2 + + +# ---------- RootModel edge cases ---------- + + +def test_optional_rootmodel_list_body(): + """Optional[RootModel[List[Model]]] — list must not be truncated.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Optional[_ItemCollection], Body()]) -> Dict[str, Any]: + return {"count": len(items.root)} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +def test_union_rootmodel_and_model(): + """Union[RootModel[List[Model]], Model] — list must not be truncated.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Union[_ItemCollection, _Item], Body()]) -> Dict[str, Any]: + if isinstance(items, _ItemCollection): + return {"type": "collection", "count": len(items.root)} + return {"type": "single", "name": items.name} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["type"] == "collection" + assert body["count"] == 3 + + +# ---------- Python 3.10+ pipe Union syntax ---------- + + +def test_pipe_union_syntax_model_or_list(): + """Model | List[Model] (PEP 604 syntax) — list must not be truncated.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[_Item | List[_Item], Body()]) -> Dict[str, Any]: # noqa: FA102 + if isinstance(items, list): + return {"count": len(items)} + return {"count": 1} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +def test_pipe_union_optional_list(): + """List[Model] | None (PEP 604 Optional) — list must not be truncated.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[List[_Item] | None, Body()]) -> Dict[str, Any]: # noqa: FA102 + if items is None: + return {"count": 0} + return {"count": len(items)} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["count"] == 3 + + +# ---------- Deeply nested: RootModel[Union[Model, List[Model]]] ---------- + + +def test_rootmodel_wrapping_union_with_sequence(): + """RootModel[Union[Model, List[Model]]] — inner Union sequence must be detected.""" + app = APIGatewayRestResolver(enable_validation=True) + + class FlexiblePayload(RootModel[Union[_Item, List[_Item]]]): + pass + + @app.post("/items") + def handler(payload: Annotated[FlexiblePayload, Body()]) -> Dict[str, Any]: + data = payload.root + if isinstance(data, list): + return {"type": "list", "count": len(data)} + return {"type": "single", "name": data.name} + + status, body = _post_json(app, "/items", _THREE_ITEMS) + assert status == 200 + assert body["type"] == "list" + assert body["count"] == 3 + + +# ---------- Multiple resolvers (ALB, HTTP API, etc.) ---------- + + +def test_union_list_body_works_across_resolvers(): + """Regression: ensure fix works for ALB and HTTP API resolvers too.""" + for ResolverClass in [APIGatewayHttpResolver, ALBResolver]: + app = ResolverClass(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]: + if isinstance(items, list): + return {"count": len(items)} + return {"count": 1} + + # Build event appropriate for resolver + if ResolverClass is APIGatewayHttpResolver: + event = load_event("apiGatewayProxyV2Event.json") + event["requestContext"]["http"]["method"] = "POST" + event["requestContext"]["http"]["path"] = "/items" + event["rawPath"] = "/items" + else: + event = load_event("albEvent.json") + event["httpMethod"] = "POST" + event["path"] = "/items" + + event["body"] = json.dumps(_THREE_ITEMS) + result = app(event, {}) + assert result["statusCode"] == 200 + body_result = json.loads(result["body"]) + assert body_result["count"] == 3, f"Failed for {ResolverClass.__name__}" + + +# ---------- Large list (stress boundary) ---------- + + +def test_union_list_body_large_payload(): + """Union[Model, List[Model]] with 100 items — no truncation.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/items") + def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]: + assert isinstance(items, list) + return {"count": len(items)} + + big_payload = [{"name": f"item-{i}", "value": i} for i in range(100)] + status, body = _post_json(app, "/items", big_payload) + assert status == 200 + assert body["count"] == 100