Skip to content

Commit 3137ecb

Browse files
danjhdleandrodamascenaclaude
authored
fix(event_handler): normalize Union and RootModel sequences in body validation (#8067)
* handle Union and RootModel sequences in Body validation * 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 <noreply@anthropic.com> --------- Co-authored-by: Leandro Damascena <lcdama@amazon.pt> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 2ab8f39 commit 3137ecb

File tree

2 files changed

+531
-3
lines changed

2 files changed

+531
-3
lines changed

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

Lines changed: 36 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,11 @@
33
import dataclasses
44
import json
55
import logging
6-
from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence, cast
6+
from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence, Union, cast
77
from urllib.parse import parse_qs
88

99
from pydantic import BaseModel
10+
from typing_extensions import get_args, get_origin
1011

1112
from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler
1213
from aws_lambda_powertools.event_handler.openapi.compat import (
@@ -25,6 +26,7 @@
2526
ResponseValidationError,
2627
)
2728
from aws_lambda_powertools.event_handler.openapi.params import Param
29+
from aws_lambda_powertools.event_handler.openapi.types import UnionType
2830

2931
if TYPE_CHECKING:
3032
from pydantic.fields import FieldInfo
@@ -431,9 +433,41 @@ def _handle_missing_field_value(
431433
values[field.name] = field.get_default()
432434

433435

436+
def _is_or_contains_sequence(annotation: Any) -> bool:
437+
"""
438+
Check if annotation is a sequence or Union/RootModel containing a sequence.
439+
440+
This function handles complex type annotations like:
441+
- List[Model] - direct sequence
442+
- Union[Model, List[Model]] - checks if any Union member is a sequence
443+
- Optional[List[Model]] - Union[List[Model], None]
444+
- RootModel[List[Model]] - checks if the RootModel wraps a sequence
445+
- Optional[RootModel[List[Model]]] - Union member that is a RootModel
446+
- RootModel[Union[Model, List[Model]]] - RootModel wrapping a Union with a sequence
447+
"""
448+
# Direct sequence check
449+
if field_annotation_is_sequence(annotation):
450+
return True
451+
452+
# Check Union members — recurse so we catch RootModel inside Union
453+
origin = get_origin(annotation)
454+
if origin is Union or origin is UnionType:
455+
for arg in get_args(annotation):
456+
if _is_or_contains_sequence(arg):
457+
return True
458+
459+
# Check if it's a RootModel wrapping a sequence (or Union containing a sequence)
460+
if lenient_issubclass(annotation, BaseModel) and getattr(annotation, "__pydantic_root_model__", False):
461+
if hasattr(annotation, "model_fields") and "root" in annotation.model_fields:
462+
root_annotation = annotation.model_fields["root"].annotation
463+
return _is_or_contains_sequence(root_annotation)
464+
465+
return False
466+
467+
434468
def _normalize_field_value(value: Any, field_info: FieldInfo) -> Any:
435469
"""Normalize field value, converting lists to single values for non-sequence fields."""
436-
if field_annotation_is_sequence(field_info.annotation):
470+
if _is_or_contains_sequence(field_info.annotation):
437471
return value
438472
elif isinstance(value, list) and value:
439473
return value[0]

0 commit comments

Comments
 (0)