feat(event_handler): generate OpenAPI specifications and validate input/output (aws-powertools#3109)

* feat: generate OpenAPI spec from event handler

* fix: resolver circular dependencies

* fix: rebase

* fix: document the new methods

* fix: linter

* fix: remove unneeded code

* fix: reduce duplication

* fix: types and sonarcube

* chore: refactor complex function

* fix: typing extensions

* fix: tests

* fix: mypy

* fix: security baseline

* feat: add simultaneous support for Pydantic v2

* fix: disable mypy and ruff on openapi compat

* chore: add explanation to imports

* chore: add first test

* fix: test

* fix: test

* fix: don't require pydantic to run normal things

* chore: added first tests

* fix: refactored tests to remove code smell

* fix: customize the handler methods

* fix: tests

* feat: add a validation middleware

* fix: uniontype

* fix: types

* fix: ignore unused-ignore

* fix: moved things around

* fix: compatibility with pydantic v2

* chore: add tests on the body request

* chore: add tests for validation middleware

* fix: assorted fixes

* fix: make tests pass in both pydantic versions

* fix: remove assert

* fix: complexity

* fix: move Response class back

* fix: more fix

* fix: more fix

* fix: one more fix

* fix: refactor OpenAPI validation middleware

* fix: refactor dependant.py

* fix: beautify encoders

* fix: move things around

* fix: costmetic changes

* fix: add more comments

* fix: format

* fix: cyclomatic

* fix: change method of generating operation id

* fix: allow validation in all resolvers

* fix: use proper resolver in tests

* fix: move from flake8 to ruff

* fix: customizing responses

* fix: add documentation to a method

* fix: more explicit comments

* fix: typo

* fix: add extra comment

* fix: comment

* fix: add comments

* fix: comments

* fix: typo

* fix: remove leftover comment

* fix: addressing comments

* fix: pydantic2 models

* fix: typing extension problems

* Adding more tests and fixing small things

* Adding more tests and fixing small things

* Adding more tests and fixing small things

* Removing flaky tests

* fix: improve coverage of encoders

* fix: mark test as pydantic v1 only

* fix: make sonarcube happy

* fix: improve coverage of params.py

* fix: add codecov.yml file to ignore compat.py

* Increasing coverage

---------

Signed-off-by: Leandro Damascena <[email protected]>
Co-authored-by: Leandro Damascena <[email protected]>
Co-authored-by: Cavalcante Damascena <[email protected]>

Author: 3 people

aws_lambda_powertools/event_handler/api_gateway.py

@@ -52,5 +52,13 @@ def __init__(
         debug: Optional[bool] = None,
         serializer: Optional[Callable[[Dict], str]] = None,
         strip_prefixes: Optional[List[Union[str, Pattern]]] = None,
+        enable_validation: bool = False,
     ):
-        super().__init__(ProxyEventType.LambdaFunctionUrlEvent, cors, debug, serializer, strip_prefixes)
+        super().__init__(
+            ProxyEventType.LambdaFunctionUrlEvent,
+            cors,
+            debug,
+            serializer,
+            strip_prefixes,
+            enable_validation,
+        )

aws_lambda_powertools/event_handler/lambda_function_url.py

@@ -0,0 +1,342 @@
+import dataclasses
+import json
+import logging
+from copy import deepcopy
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.event_handler import Response
+from aws_lambda_powertools.event_handler.api_gateway import Route
+from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler, NextMiddleware
+from aws_lambda_powertools.event_handler.openapi.compat import (
+    ModelField,
+    _model_dump,
+    _normalize_errors,
+    _regenerate_error_with_loc,
+    get_missing_field_error,
+)
+from aws_lambda_powertools.event_handler.openapi.encoders import jsonable_encoder
+from aws_lambda_powertools.event_handler.openapi.exceptions import RequestValidationError
+from aws_lambda_powertools.event_handler.openapi.params import Param
+from aws_lambda_powertools.event_handler.openapi.types import IncEx
+from aws_lambda_powertools.event_handler.types import EventHandlerInstance
+
+logger = logging.getLogger(__name__)
+
+
+class OpenAPIValidationMiddleware(BaseMiddlewareHandler):
+    """
+    OpenAPIValidationMiddleware is a middleware that validates the request against the OpenAPI schema defined by the
+    Lambda handler. It also validates the response against the OpenAPI schema defined by the Lambda handler. It
+    should not be used directly, but rather through the `enable_validation` parameter of the `ApiGatewayResolver`.
+
+    Examples
+    --------
+
+    ```python
+    from typing import List
+
+    from pydantic import BaseModel
+
+    from aws_lambda_powertools.event_handler.api_gateway import (
+        APIGatewayRestResolver,
+    )
+
+    class Todo(BaseModel):
+      name: str
+
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/todos")
+    def get_todos(): List[Todo]:
+      return [Todo(name="hello world")]
+    ```
+    """
+
+    def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) -> Response:
+        logger.debug("OpenAPIValidationMiddleware handler")
+
+        route: Route = app.context["_route"]
+
+        values: Dict[str, Any] = {}
+        errors: List[Any] = []
+
+        try:
+            # Process path values, which can be found on the route_args
+            path_values, path_errors = _request_params_to_args(
+                route.dependant.path_params,
+                app.context["_route_args"],
+            )
+
+            # Process query values
+            query_values, query_errors = _request_params_to_args(
+                route.dependant.query_params,
+                app.current_event.query_string_parameters or {},
+            )
+
+            values.update(path_values)
+            values.update(query_values)
+            errors += path_errors + query_errors
+
+            # Process the request body, if it exists
+            if route.dependant.body_params:
+                (body_values, body_errors) = _request_body_to_args(
+                    required_params=route.dependant.body_params,
+                    received_body=self._get_body(app),
+                )
+                values.update(body_values)
+                errors.extend(body_errors)
+
+            if errors:
+                # Raise the validation errors
+                raise RequestValidationError(_normalize_errors(errors))
+            else:
+                # Re-write the route_args with the validated values, and call the next middleware
+                app.context["_route_args"] = values
+                response = next_middleware(app)
+
+                # Process the response body if it exists
+                raw_response = jsonable_encoder(response.body)
+
+                # Validate and serialize the response
+                return self._serialize_response(field=route.dependant.return_param, response_content=raw_response)
+        except RequestValidationError as e:
+            return Response(
+                status_code=422,
+                content_type="application/json",
+                body=json.dumps({"detail": e.errors()}),
+            )
+
+    def _serialize_response(
+        self,
+        *,
+        field: Optional[ModelField] = None,
+        response_content: Any,
+        include: Optional[IncEx] = None,
+        exclude: Optional[IncEx] = None,
+        by_alias: bool = True,
+        exclude_unset: bool = False,
+        exclude_defaults: bool = False,
+        exclude_none: bool = False,
+    ) -> Any:
+        """
+        Serialize the response content according to the field type.
+        """
+        if field:
+            errors: List[Dict[str, Any]] = []
+            # MAINTENANCE: remove this when we drop pydantic v1
+            if not hasattr(field, "serializable"):
+                response_content = self._prepare_response_content(
+                    response_content,
+                    exclude_unset=exclude_unset,
+                    exclude_defaults=exclude_defaults,
+                    exclude_none=exclude_none,
+                )
+
+            value = _validate_field(field=field, value=response_content, loc=("response",), existing_errors=errors)
+            if errors:
+                raise RequestValidationError(errors=_normalize_errors(errors), body=response_content)
+
+            if hasattr(field, "serialize"):
+                return field.serialize(
+                    value,
+                    include=include,
+                    exclude=exclude,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_defaults=exclude_defaults,
+                    exclude_none=exclude_none,
+                )
+
+            return jsonable_encoder(
+                value,
+                include=include,
+                exclude=exclude,
+                by_alias=by_alias,
+                exclude_unset=exclude_unset,
+                exclude_defaults=exclude_defaults,
+                exclude_none=exclude_none,
+            )
+        else:
+            # Just serialize the response content returned from the handler
+            return jsonable_encoder(response_content)
+
+    def _prepare_response_content(
+        self,
+        res: Any,
+        *,
+        exclude_unset: bool,
+        exclude_defaults: bool = False,
+        exclude_none: bool = False,
+    ) -> Any:
+        """
+        Prepares the response content for serialization.
+        """
+        if isinstance(res, BaseModel):
+            return _model_dump(
+                res,
+                by_alias=True,
+                exclude_unset=exclude_unset,
+                exclude_defaults=exclude_defaults,
+                exclude_none=exclude_none,
+            )
+        elif isinstance(res, list):
+            return [
+                self._prepare_response_content(item, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults)
+                for item in res
+            ]
+        elif isinstance(res, dict):
+            return {
+                k: self._prepare_response_content(v, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults)
+                for k, v in res.items()
+            }
+        elif dataclasses.is_dataclass(res):
+            return dataclasses.asdict(res)
+        return res
+
+    def _get_body(self, app: EventHandlerInstance) -> Dict[str, Any]:
+        """
+        Get the request body from the event, and parse it as JSON.
+        """
+
+        content_type_value = app.current_event.get_header_value("content-type")
+        if not content_type_value or content_type_value.startswith("application/json"):
+            try:
+                return app.current_event.json_body
+            except json.JSONDecodeError as e:
+                raise RequestValidationError(
+                    [
+                        {
+                            "type": "json_invalid",
+                            "loc": ("body", e.pos),
+                            "msg": "JSON decode error",
+                            "input": {},
+                            "ctx": {"error": e.msg},
+                        },
+                    ],
+                    body=e.doc,
+                ) from e
+        else:
+            raise NotImplementedError("Only JSON body is supported")
+
+
+def _request_params_to_args(
+    required_params: Sequence[ModelField],
+    received_params: Mapping[str, Any],
+) -> Tuple[Dict[str, Any], List[Any]]:
+    """
+    Convert the request params to a dictionary of values using validation, and returns a list of errors.
+    """
+    values = {}
+    errors = []
+
+    for field in required_params:
+        value = received_params.get(field.alias)
+
+        field_info = field.field_info
+        if not isinstance(field_info, Param):
+            raise AssertionError(f"Expected Param field_info, got {field_info}")
+
+        loc = (field_info.in_.value, field.alias)
+
+        # If we don't have a value, see if it's required or has a default
+        if value is None:
+            if field.required:
+                errors.append(get_missing_field_error(loc=loc))
+            else:
+                values[field.name] = deepcopy(field.default)
+            continue
+
+        # Finally, validate the value
+        values[field.name] = _validate_field(field=field, value=value, loc=loc, existing_errors=errors)
+
+    return values, errors
+
+
+def _request_body_to_args(
+    required_params: List[ModelField],
+    received_body: Optional[Dict[str, Any]],
+) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
+    """
+    Convert the request body to a dictionary of values using validation, and returns a list of errors.
+    """
+    values: Dict[str, Any] = {}
+    errors: List[Dict[str, Any]] = []
+
+    received_body, field_alias_omitted = _get_embed_body(
+        field=required_params[0],
+        required_params=required_params,
+        received_body=received_body,
+    )
+
+    for field in required_params:
+        # This sets the location to:
+        # { "user": { object } } if field.alias == user
+        # { { object } if field_alias is omitted
+        loc: Tuple[str, ...] = ("body", field.alias)
+        if field_alias_omitted:
+            loc = ("body",)
+
+        value: Optional[Any] = None
+
+        # Now that we know what to look for, try to get the value from the received body
+        if received_body is not None:
+            try:
+                value = received_body.get(field.alias)
+            except AttributeError:
+                errors.append(get_missing_field_error(loc))
+                continue
+
+        # Determine if the field is required
+        if value is None:
+            if field.required:
+                errors.append(get_missing_field_error(loc))
+            else:
+                values[field.name] = deepcopy(field.default)
+            continue
+
+        # MAINTENANCE: Handle byte and file fields
+
+        # Finally, validate the value
+        values[field.name] = _validate_field(field=field, value=value, loc=loc, existing_errors=errors)
+
+    return values, errors
+
+
+def _validate_field(
+    *,
+    field: ModelField,
+    value: Any,
+    loc: Tuple[str, ...],
+    existing_errors: List[Dict[str, Any]],
+):
+    """
+    Validate a field, and append any errors to the existing_errors list.
+    """
+    validated_value, errors = field.validate(value, value, loc=loc)
+
+    if isinstance(errors, list):
+        processed_errors = _regenerate_error_with_loc(errors=errors, loc_prefix=())
+        existing_errors.extend(processed_errors)
+    elif errors:
+        existing_errors.append(errors)
+
+    return validated_value
+
+
+def _get_embed_body(
+    *,
+    field: ModelField,
+    required_params: List[ModelField],
+    received_body: Optional[Dict[str, Any]],
+) -> Tuple[Optional[Dict[str, Any]], bool]:
+    field_info = field.field_info
+    embed = getattr(field_info, "embed", None)
+
+    # If the field is an embed, and the field alias is omitted, we need to wrap the received body in the field alias.
+    field_alias_omitted = len(required_params) == 1 and not embed
+    if field_alias_omitted:
+        received_body = {field.alias: received_body}
+
+    return received_body, field_alias_omitted

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -0,0 +1,342 @@
+import inspect
+import re
+from typing import Any, Callable, Dict, ForwardRef, List, Optional, Set, Tuple, Type, cast
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.event_handler.openapi.compat import (
+    ModelField,
+    create_body_model,
+    evaluate_forwardref,
+    is_scalar_field,
+    is_scalar_sequence_field,
+)
+from aws_lambda_powertools.event_handler.openapi.params import (
+    Body,
+    Dependant,
+    File,
+    Form,
+    Header,
+    Param,
+    ParamTypes,
+    Query,
+    analyze_param,
+    create_response_field,
+    get_flat_dependant,
+)
+
+"""
+This turns the opaque function signature into typed, validated models.
+
+It relies on Pydantic's typing and validation to achieve this in a declarative way.
+This enables traits like autocompletion, validation, and declarative structure vs imperative parsing.
+
+This code parses an OpenAPI operation handler function signature into Pydantic models. It uses inspect to get the
+signature and regex to parse path parameters. Each parameter is analyzed to extract its type annotation and generate
+a corresponding Pydantic field, which are added to a Dependant model. Return values are handled similarly.
+
+This modeling allows for type checking, automatic parameter name/location/type extraction, and input validation -
+turning the opaque signature into validated models. It relies on Pydantic's typing and validation for a declarative
+approach over imperative parsing, enabling autocompletion, validation and structure.
+"""
+
+
+def add_param_to_fields(
+    *,
+    field: ModelField,
+    dependant: Dependant,
+) -> None:
+    """
+    Adds a parameter to the list of parameters in the dependant model.
+
+    Parameters
+    ----------
+    field: ModelField
+        The field to add
+    dependant: Dependant
+        The dependant model to add the field to
+
+    """
+    field_info = cast(Param, field.field_info)
+    if field_info.in_ == ParamTypes.path:
+        dependant.path_params.append(field)
+    elif field_info.in_ == ParamTypes.query:
+        dependant.query_params.append(field)
+    elif field_info.in_ == ParamTypes.header:
+        dependant.header_params.append(field)
+    else:
+        if field_info.in_ != ParamTypes.cookie:
+            raise AssertionError(f"Unsupported param type: {field_info.in_}")
+        dependant.cookie_params.append(field)
+
+
+def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
+    """
+    Evaluates a type annotation, which can be a string or a ForwardRef.
+    """
+    if isinstance(annotation, str):
+        annotation = ForwardRef(annotation)
+        annotation = evaluate_forwardref(annotation, globalns, globalns)
+    return annotation
+
+
+def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
+    """
+    Returns a typed signature for a callable, resolving forward references.
+
+    Parameters
+    ----------
+    call: Callable[..., Any]
+        The callable to get the signature for
+
+    Returns
+    -------
+    inspect.Signature
+        The typed signature
+    """
+    signature = inspect.signature(call)
+
+    # Gets the global namespace for the call. This is used to resolve forward references.
+    globalns = getattr(call, "__global__", {})
+
+    typed_params = [
+        inspect.Parameter(
+            name=param.name,
+            kind=param.kind,
+            default=param.default,
+            annotation=get_typed_annotation(param.annotation, globalns),
+        )
+        for param in signature.parameters.values()
+    ]
+
+    # If the return annotation is not empty, add it to the signature.
+    if signature.return_annotation is not inspect.Signature.empty:
+        return_param = inspect.Parameter(
+            name="Return",
+            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            default=None,
+            annotation=get_typed_annotation(signature.return_annotation, globalns),
+        )
+        return inspect.Signature(typed_params, return_annotation=return_param.annotation)
+    else:
+        return inspect.Signature(typed_params)
+
+
+def get_path_param_names(path: str) -> Set[str]:
+    """
+    Returns the path parameter names from a path template. Those are the strings between < and >.
+
+    Parameters
+    ----------
+    path: str
+        The path template
+
+    Returns
+    -------
+    Set[str]
+        The path parameter names
+
+    """
+    return set(re.findall("<(.*?)>", path))
+
+
+def get_dependant(
+    *,
+    path: str,
+    call: Callable[..., Any],
+    name: Optional[str] = None,
+) -> Dependant:
+    """
+    Returns a dependant model for a handler function. A dependant model is a model that contains
+    the parameters and return value of a handler function.
+
+    Parameters
+    ----------
+    path: str
+        The path template
+    call: Callable[..., Any]
+        The handler function
+    name: str, optional
+        The name of the handler function
+
+    Returns
+    -------
+    Dependant
+        The dependant model for the handler function
+    """
+    path_param_names = get_path_param_names(path)
+    endpoint_signature = get_typed_signature(call)
+    signature_params = endpoint_signature.parameters
+
+    dependant = Dependant(
+        call=call,
+        name=name,
+        path=path,
+    )
+
+    # Add each parameter to the dependant model
+    for param_name, param in signature_params.items():
+        # If the parameter is a path parameter, we need to set the in_ field to "path".
+        is_path_param = param_name in path_param_names
+
+        # Analyze the parameter to get the Pydantic field.
+        param_field = analyze_param(
+            param_name=param_name,
+            annotation=param.annotation,
+            value=param.default,
+            is_path_param=is_path_param,
+            is_response_param=False,
+        )
+        if param_field is None:
+            raise AssertionError(f"Parameter field is None for param: {param_name}")
+
+        if is_body_param(param_field=param_field, is_path_param=is_path_param):
+            dependant.body_params.append(param_field)
+        else:
+            add_param_to_fields(field=param_field, dependant=dependant)
+
+    # If the return annotation is not empty, add it to the dependant model.
+    return_annotation = endpoint_signature.return_annotation
+    if return_annotation is not inspect.Signature.empty:
+        param_field = analyze_param(
+            param_name="return",
+            annotation=return_annotation,
+            value=None,
+            is_path_param=False,
+            is_response_param=True,
+        )
+        if param_field is None:
+            raise AssertionError("Param field is None for return annotation")
+
+        dependant.return_param = param_field
+
+    return dependant
+
+
+def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
+    """
+    Returns whether a parameter is a request body parameter, by checking if it is a scalar field or a body field.
+
+    Parameters
+    ----------
+    param_field: ModelField
+        The parameter field
+    is_path_param: bool
+        Whether the parameter is a path parameter
+
+    Returns
+    -------
+    bool
+        Whether the parameter is a request body parameter
+    """
+    if is_path_param:
+        if not is_scalar_field(field=param_field):
+            raise AssertionError("Path params must be of one of the supported types")
+        return False
+    elif is_scalar_field(field=param_field):
+        return False
+    elif isinstance(param_field.field_info, (Query, Header)) and is_scalar_sequence_field(param_field):
+        return False
+    else:
+        if not isinstance(param_field.field_info, Body):
+            raise AssertionError(f"Param: {param_field.name} can only be a request body, use Body()")
+        return True
+
+
+def get_flat_params(dependant: Dependant) -> List[ModelField]:
+    """
+    Get a list of all the parameters from a Dependant object.
+
+    Parameters
+    ----------
+    dependant : Dependant
+        The Dependant object containing the parameters.
+
+    Returns
+    -------
+    List[ModelField]
+        A list of ModelField objects containing the flat parameters from the Dependant object.
+
+    """
+    flat_dependant = get_flat_dependant(dependant)
+    return (
+        flat_dependant.path_params
+        + flat_dependant.query_params
+        + flat_dependant.header_params
+        + flat_dependant.cookie_params
+    )
+
+
+def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]:
+    """
+    Get the Body field for a given Dependant object.
+    """
+
+    flat_dependant = get_flat_dependant(dependant)
+    if not flat_dependant.body_params:
+        return None
+
+    first_param = flat_dependant.body_params[0]
+    field_info = first_param.field_info
+
+    # Handle the case where there is only one body parameter and it is embedded
+    embed = getattr(field_info, "embed", None)
+    body_param_names_set = {param.name for param in flat_dependant.body_params}
+    if len(body_param_names_set) == 1 and not embed:
+        return first_param
+
+    # If one field requires to embed, all have to be embedded
+    for param in flat_dependant.body_params:
+        setattr(param.field_info, "embed", True)  # noqa: B010
+
+    # Generate a custom body model for this endpoint
+    model_name = "Body_" + name
+    body_model = create_body_model(fields=flat_dependant.body_params, model_name=model_name)
+
+    required = any(True for f in flat_dependant.body_params if f.required)
+
+    body_field_info, body_field_info_kwargs = get_body_field_info(
+        body_model=body_model,
+        flat_dependant=flat_dependant,
+        required=required,
+    )
+
+    final_field = create_response_field(
+        name="body",
+        type_=body_model,
+        required=required,
+        alias="body",
+        field_info=body_field_info(**body_field_info_kwargs),
+    )
+    return final_field
+
+
+def get_body_field_info(
+    *,
+    body_model: Type[BaseModel],
+    flat_dependant: Dependant,
+    required: bool,
+) -> Tuple[Type[Body], Dict[str, Any]]:
+    """
+    Get the Body field info and kwargs for a given body model.
+    """
+
+    body_field_info_kwargs: Dict[str, Any] = {"annotation": body_model, "alias": "body"}
+
+    if not required:
+        body_field_info_kwargs["default"] = None
+
+    if any(isinstance(f.field_info, File) for f in flat_dependant.body_params):
+        body_field_info: Type[Body] = File
+    elif any(isinstance(f.field_info, Form) for f in flat_dependant.body_params):
+        body_field_info = Form
+    else:
+        body_field_info = Body
+
+        body_param_media_types = [
+            f.field_info.media_type for f in flat_dependant.body_params if isinstance(f.field_info, Body)
+        ]
+        if len(set(body_param_media_types)) == 1:
+            body_field_info_kwargs["media_type"] = body_param_media_types[0]
+
+    return body_field_info, body_field_info_kwargs

aws_lambda_powertools/event_handler/openapi/__init__.py

@@ -0,0 +1,344 @@
+import dataclasses
+import datetime
+from collections import defaultdict, deque
+from decimal import Decimal
+from enum import Enum
+from pathlib import Path, PurePath
+from re import Pattern
+from types import GeneratorType
+from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
+from uuid import UUID
+
+from pydantic import BaseModel
+from pydantic.color import Color
+from pydantic.types import SecretBytes, SecretStr
+
+from aws_lambda_powertools.event_handler.openapi.compat import _model_dump
+from aws_lambda_powertools.event_handler.openapi.types import IncEx
+
+"""
+This module contains the encoders used by jsonable_encoder to convert Python objects to JSON serializable data types.
+"""
+
+
+def jsonable_encoder(  # noqa: PLR0911
+    obj: Any,
+    include: Optional[IncEx] = None,
+    exclude: Optional[IncEx] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_defaults: bool = False,
+    exclude_none: bool = False,
+) -> Any:
+    """
+    JSON encodes an arbitrary Python object into JSON serializable data types.
+
+    This is a modified version of fastapi.encoders.jsonable_encoder that supports
+    encoding of pydantic.BaseModel objects.
+
+    Parameters
+    ----------
+    obj : Any
+        The object to encode
+    include : Optional[IncEx], optional
+        A set or dictionary of strings that specifies which properties should be included, by default None,
+        meaning everything is included
+    exclude : Optional[IncEx], optional
+        A set or dictionary of strings that specifies which properties should be excluded, by default None,
+        meaning nothing is excluded
+    by_alias : bool, optional
+        Whether field aliases should be respected, by default True
+    exclude_unset : bool, optional
+        Whether fields that are not set should be excluded, by default False
+    exclude_defaults : bool, optional
+        Whether fields that are equal to their default value (as specified in the model) should be excluded,
+        by default False
+    exclude_none : bool, optional
+        Whether fields that are equal to None should be excluded, by default False
+
+    Returns
+    -------
+    Any
+        The JSON serializable data types
+    """
+    if include is not None and not isinstance(include, (set, dict)):
+        include = set(include)
+    if exclude is not None and not isinstance(exclude, (set, dict)):
+        exclude = set(exclude)
+
+    # Pydantic models
+    if isinstance(obj, BaseModel):
+        return _dump_base_model(
+            obj=obj,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+        )
+
+    # Dataclasses
+    if dataclasses.is_dataclass(obj):
+        obj_dict = dataclasses.asdict(obj)
+        return jsonable_encoder(
+            obj_dict,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_defaults=exclude_defaults,
+            exclude_none=exclude_none,
+        )
+
+    # Enums
+    if isinstance(obj, Enum):
+        return obj.value
+
+    # Paths
+    if isinstance(obj, PurePath):
+        return str(obj)
+
+    # Scalars
+    if isinstance(obj, (str, int, float, type(None))):
+        return obj
+
+    # Dictionaries
+    if isinstance(obj, dict):
+        return _dump_dict(
+            obj=obj,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_none=exclude_none,
+            exclude_unset=exclude_unset,
+        )
+
+    # Sequences
+    if isinstance(obj, (list, set, frozenset, GeneratorType, tuple, deque)):
+        return _dump_sequence(
+            obj=obj,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+            exclude_unset=exclude_unset,
+        )
+
+    # Other types
+    if type(obj) in ENCODERS_BY_TYPE:
+        return ENCODERS_BY_TYPE[type(obj)](obj)
+
+    for encoder, classes_tuple in encoders_by_class_tuples.items():
+        if isinstance(obj, classes_tuple):
+            return encoder(obj)
+
+    # Default
+    return _dump_other(
+        obj=obj,
+        include=include,
+        exclude=exclude,
+        by_alias=by_alias,
+        exclude_none=exclude_none,
+        exclude_unset=exclude_unset,
+        exclude_defaults=exclude_defaults,
+    )
+
+
+def _dump_base_model(
+    *,
+    obj: Any,
+    include: Optional[IncEx] = None,
+    exclude: Optional[IncEx] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_none: bool = False,
+    exclude_defaults: bool = False,
+):
+    """
+    Dump a BaseModel object to a dict, using the same parameters as jsonable_encoder
+    """
+    obj_dict = _model_dump(
+        obj,
+        mode="json",
+        include=include,
+        exclude=exclude,
+        by_alias=by_alias,
+        exclude_unset=exclude_unset,
+        exclude_none=exclude_none,
+        exclude_defaults=exclude_defaults,
+    )
+    if "__root__" in obj_dict:
+        obj_dict = obj_dict["__root__"]
+
+    return jsonable_encoder(
+        obj_dict,
+        exclude_none=exclude_none,
+        exclude_defaults=exclude_defaults,
+    )
+
+
+def _dump_dict(
+    *,
+    obj: Any,
+    include: Optional[IncEx] = None,
+    exclude: Optional[IncEx] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_none: bool = False,
+) -> Dict[str, Any]:
+    """
+    Dump a dict to a dict, using the same parameters as jsonable_encoder
+    """
+    encoded_dict = {}
+    allowed_keys = set(obj.keys())
+    if include is not None:
+        allowed_keys &= set(include)
+    if exclude is not None:
+        allowed_keys -= set(exclude)
+    for key, value in obj.items():
+        if (
+            (not isinstance(key, str) or not key.startswith("_sa"))
+            and (value is not None or not exclude_none)
+            and key in allowed_keys
+        ):
+            encoded_key = jsonable_encoder(
+                key,
+                by_alias=by_alias,
+                exclude_unset=exclude_unset,
+                exclude_none=exclude_none,
+            )
+            encoded_value = jsonable_encoder(
+                value,
+                by_alias=by_alias,
+                exclude_unset=exclude_unset,
+                exclude_none=exclude_none,
+            )
+            encoded_dict[encoded_key] = encoded_value
+    return encoded_dict
+
+
+def _dump_sequence(
+    *,
+    obj: Any,
+    include: Optional[IncEx] = None,
+    exclude: Optional[IncEx] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_none: bool = False,
+    exclude_defaults: bool = False,
+) -> List[Any]:
+    """
+    Dump a sequence to a list, using the same parameters as jsonable_encoder
+    """
+    encoded_list = []
+    for item in obj:
+        encoded_list.append(
+            jsonable_encoder(
+                item,
+                include=include,
+                exclude=exclude,
+                by_alias=by_alias,
+                exclude_unset=exclude_unset,
+                exclude_defaults=exclude_defaults,
+                exclude_none=exclude_none,
+            ),
+        )
+    return encoded_list
+
+
+def _dump_other(
+    *,
+    obj: Any,
+    include: Optional[IncEx] = None,
+    exclude: Optional[IncEx] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_none: bool = False,
+    exclude_defaults: bool = False,
+) -> Any:
+    """
+    Dump an object to ah hashable object, using the same parameters as jsonable_encoder
+    """
+    try:
+        data = dict(obj)
+    except Exception as e:
+        errors: List[Exception] = [e]
+        try:
+            data = vars(obj)
+        except Exception as e:
+            errors.append(e)
+            raise ValueError(errors) from e
+    return jsonable_encoder(
+        data,
+        include=include,
+        exclude=exclude,
+        by_alias=by_alias,
+        exclude_unset=exclude_unset,
+        exclude_defaults=exclude_defaults,
+        exclude_none=exclude_none,
+    )
+
+
+def iso_format(o: Union[datetime.date, datetime.time]) -> str:
+    """
+    ISO format for date and time
+    """
+    return o.isoformat()
+
+
+def decimal_encoder(dec_value: Decimal) -> Union[int, float]:
+    """
+    Encodes a Decimal as int of there's no exponent, otherwise float
+
+    This is useful when we use ConstrainedDecimal to represent Numeric(x,0)
+    where an integer (but not int typed) is used. Encoding this as a float
+    results in failed round-tripping between encode and parse.
+
+    >>> decimal_encoder(Decimal("1.0"))
+    1.0
+
+    >>> decimal_encoder(Decimal("1"))
+    1
+    """
+    if dec_value.as_tuple().exponent >= 0:  # type: ignore[operator]
+        return int(dec_value)
+    else:
+        return float(dec_value)
+
+
+# Encoders for types that are not JSON serializable
+ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
+    bytes: lambda o: o.decode(),
+    Color: str,
+    datetime.date: iso_format,
+    datetime.datetime: iso_format,
+    datetime.time: iso_format,
+    datetime.timedelta: lambda td: td.total_seconds(),
+    Decimal: decimal_encoder,
+    Enum: lambda o: o.value,
+    frozenset: list,
+    deque: list,
+    GeneratorType: list,
+    Path: str,
+    Pattern: lambda o: o.pattern,
+    SecretBytes: str,
+    SecretStr: str,
+    set: list,
+    UUID: str,
+}
+
+
+# Generates a mapping of encoders to a tuple of classes that they can encode
+def generate_encoders_by_class_tuples(
+    type_encoder_map: Dict[Any, Callable[[Any], Any]],
+) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]:
+    encoders: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(tuple)
+    for type_, encoder in type_encoder_map.items():
+        encoders[encoder] += (type_,)
+    return encoders
+
+
+# Mapping of encoders to a tuple of classes that they can encode
+encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)

aws_lambda_powertools/event_handler/openapi/compat.py

@@ -0,0 +1,23 @@
+from typing import Any, Sequence
+
+
+class ValidationException(Exception):
+    """
+    Base exception for all validation errors
+    """
+
+    def __init__(self, errors: Sequence[Any]) -> None:
+        self._errors = errors
+
+    def errors(self) -> Sequence[Any]:
+        return self._errors
+
+
+class RequestValidationError(ValidationException):
+    """
+    Raised when the request body does not match the OpenAPI schema
+    """
+
+    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
+        super().__init__(errors)
+        self.body = body

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -0,0 +1,52 @@
+import types
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Set, Type, Union
+
+if TYPE_CHECKING:
+    from pydantic import BaseModel  # noqa: F401
+
+CacheKey = Optional[Callable[..., Any]]
+IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any]]
+ModelNameMap = Dict[Union[Type["BaseModel"], Type[Enum]], str]
+TypeModelOrEnum = Union[Type["BaseModel"], Type[Enum]]
+UnionType = getattr(types, "UnionType", Union)
+
+
+COMPONENT_REF_PREFIX = "#/components/schemas/"
+COMPONENT_REF_TEMPLATE = "#/components/schemas/{model}"
+METHODS_WITH_BODY = {"GET", "HEAD", "POST", "PUT", "DELETE", "PATCH"}
+
+try:
+    from pydantic.version import VERSION as PYDANTIC_VERSION
+
+    PYDANTIC_V2 = PYDANTIC_VERSION.startswith("2.")
+except ImportError:
+    PYDANTIC_V2 = False
+
+
+validation_error_definition = {
+    "title": "ValidationError",
+    "type": "object",
+    "properties": {
+        "loc": {
+            "title": "Location",
+            "type": "array",
+            "items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
+        },
+        "msg": {"title": "Message", "type": "string"},
+        "type": {"title": "Error Type", "type": "string"},
+    },
+    "required": ["loc", "msg", "type"],
+}
+
+validation_error_response_definition = {
+    "title": "HTTPValidationError",
+    "type": "object",
+    "properties": {
+        "detail": {
+            "title": "Detail",
+            "type": "array",
+            "items": {"$ref": COMPONENT_REF_PREFIX + "ValidationError"},
+        },
+    },
+}

aws_lambda_powertools/event_handler/openapi/encoders.py

@@ -48,9 +48,10 @@ def __init__(
         debug: Optional[bool] = None,
         serializer: Optional[Callable[[Dict], str]] = None,
         strip_prefixes: Optional[List[Union[str, Pattern]]] = None,
+        enable_validation: bool = False,
     ):
         """Amazon VPC Lattice resolver"""
-        super().__init__(ProxyEventType.VPCLatticeEvent, cors, debug, serializer, strip_prefixes)
+        super().__init__(ProxyEventType.VPCLatticeEvent, cors, debug, serializer, strip_prefixes, enable_validation)
 
 
 class VPCLatticeV2Resolver(ApiGatewayResolver):
@@ -93,6 +94,7 @@ def __init__(
         debug: Optional[bool] = None,
         serializer: Optional[Callable[[Dict], str]] = None,
         strip_prefixes: Optional[List[Union[str, Pattern]]] = None,
+        enable_validation: bool = False,
     ):
         """Amazon VPC Lattice resolver"""
-        super().__init__(ProxyEventType.VPCLatticeEventV2, cors, debug, serializer, strip_prefixes)
+        super().__init__(ProxyEventType.VPCLatticeEventV2, cors, debug, serializer, strip_prefixes, enable_validation)

aws_lambda_powertools/event_handler/openapi/exceptions.py

@@ -6,20 +6,26 @@
 else:
     from typing_extensions import Literal, Protocol, TypedDict
 
+if sys.version_info >= (3, 9):
+    from typing import Annotated
+else:
+    from typing_extensions import Annotated
 
 if sys.version_info >= (3, 11):
     from typing import NotRequired
 else:
     from typing_extensions import NotRequired
 
 
+# Even though `get_args` and `get_origin` were added in Python 3.8, they only handle Annotated correctly on 3.10.
+# So for python < 3.10 we use the backport from typing_extensions.
 if sys.version_info >= (3, 10):
-    from typing import TypeAlias
+    from typing import TypeAlias, get_args, get_origin
 else:
-    from typing_extensions import TypeAlias
+    from typing_extensions import TypeAlias, get_args, get_origin
 
 AnyCallableT = TypeVar("AnyCallableT", bound=Callable[..., Any])  # noqa: VNE001
 # JSON primitives only, mypy doesn't support recursive tho
 JSONType = Union[str, int, float, bool, None, Dict[str, Any], List[Any]]
 
-__all__ = ["Protocol", "TypedDict", "Literal", "NotRequired", "TypeAlias"]
+__all__ = ["get_args", "get_origin", "Annotated", "Protocol", "TypedDict", "Literal", "NotRequired", "TypeAlias"]

aws_lambda_powertools/event_handler/openapi/models.py

@@ -0,0 +1,2 @@
+ignore:
+  - "aws_lambda_powertools/event_handler/openapi/compat.py"

aws_lambda_powertools/event_handler/openapi/params.py

@@ -169,6 +169,7 @@ exclude = '''
     | buck-out
     | build
     | dist
+    | aws_lambda_powertools/event_handler/openapi/compat.py
   )/
   | example
 )

aws_lambda_powertools/event_handler/openapi/types.py

@@ -87,5 +87,6 @@ split-on-trailing-comma = true
 "tests/e2e/utils/data_fetcher/__init__.py" = ["F401"]
 "aws_lambda_powertools/utilities/data_classes/s3_event.py" = ["A003"]
 "aws_lambda_powertools/utilities/parser/models/__init__.py" = ["E402"]
+"aws_lambda_powertools/event_handler/openapi/compat.py" = ["F401"]
 # Maintenance: we're keeping EphemeralMetrics code in case of Hyrum's law so we can quickly revert it
 "aws_lambda_powertools/metrics/metrics.py" = ["ERA001"]

aws_lambda_powertools/event_handler/vpc_lattice.py

@@ -0,0 +1,195 @@
+import math
+from dataclasses import dataclass
+from typing import List
+
+import pytest
+from pydantic import BaseModel
+from pydantic.color import Color
+
+from aws_lambda_powertools.event_handler.openapi.encoders import jsonable_encoder
+
+
+@pytest.fixture
+def pydanticv1_only():
+    from pydantic import __version__
+
+    version = __version__.split(".")
+    if version[0] != "1":
+        pytest.skip("pydanticv1 test only")
+
+
+def test_openapi_encode_include():
+    class User(BaseModel):
+        name: str
+        age: int
+
+    result = jsonable_encoder(User(name="John", age=20), include=["name"])
+    assert result == {"name": "John"}
+
+
+def test_openapi_encode_exclude():
+    class User(BaseModel):
+        name: str
+        age: int
+
+    result = jsonable_encoder(User(name="John", age=20), exclude=["age"])
+    assert result == {"name": "John"}
+
+
+def test_openapi_encode_pydantic():
+    class Order(BaseModel):
+        quantity: int
+
+    class User(BaseModel):
+        name: str
+        order: Order
+
+    result = jsonable_encoder(User(name="John", order=Order(quantity=2)))
+    assert result == {"name": "John", "order": {"quantity": 2}}
+
+
+@pytest.mark.usefixtures("pydanticv1_only")
+def test_openapi_encode_pydantic_root_types():
+    class User(BaseModel):
+        __root__: List[str]
+
+    result = jsonable_encoder(User(__root__=["John", "Jane"]))
+    assert result == ["John", "Jane"]
+
+
+def test_openapi_encode_dataclass():
+    @dataclass
+    class Order:
+        quantity: int
+
+    @dataclass
+    class User:
+        name: str
+        order: Order
+
+    result = jsonable_encoder(User(name="John", order=Order(quantity=2)))
+    assert result == {"name": "John", "order": {"quantity": 2}}
+
+
+def test_openapi_encode_enum():
+    from enum import Enum
+
+    class Color(Enum):
+        RED = "red"
+        GREEN = "green"
+        BLUE = "blue"
+
+    result = jsonable_encoder(Color.RED)
+    assert result == "red"
+
+
+def test_openapi_encode_purepath():
+    from pathlib import PurePath
+
+    result = jsonable_encoder(PurePath("/foo/bar"))
+    assert result == "/foo/bar"
+
+
+def test_openapi_encode_scalars():
+    result = jsonable_encoder("foo")
+    assert result == "foo"
+
+    result = jsonable_encoder(1)
+    assert result == 1
+
+    result = jsonable_encoder(1.0)
+    assert math.isclose(result, 1.0)
+
+    result = jsonable_encoder(True)
+    assert result is True
+
+    result = jsonable_encoder(None)
+    assert result is None
+
+
+def test_openapi_encode_dict():
+    result = jsonable_encoder({"foo": "bar"})
+    assert result == {"foo": "bar"}
+
+
+def test_openapi_encode_dict_with_include():
+    result = jsonable_encoder({"foo": "bar", "bar": "foo"}, include=["foo"])
+    assert result == {"foo": "bar"}
+
+
+def test_openapi_encode_dict_with_exclude():
+    result = jsonable_encoder({"foo": "bar", "bar": "foo"}, exclude=["bar"])
+    assert result == {"foo": "bar"}
+
+
+def test_openapi_encode_sequences():
+    result = jsonable_encoder(["foo", "bar"])
+    assert result == ["foo", "bar"]
+
+    result = jsonable_encoder(("foo", "bar"))
+    assert result == ["foo", "bar"]
+
+    result = jsonable_encoder({"foo", "bar"})
+    assert set(result) == {"foo", "bar"}
+
+    result = jsonable_encoder(frozenset(("foo", "bar")))
+    assert set(result) == {"foo", "bar"}
+
+
+def test_openapi_encode_bytes():
+    result = jsonable_encoder(b"foo")
+    assert result == "foo"
+
+
+def test_openapi_encode_timedelta():
+    from datetime import timedelta
+
+    result = jsonable_encoder(timedelta(seconds=1))
+    assert result == 1
+
+
+def test_openapi_encode_decimal():
+    from decimal import Decimal
+
+    result = jsonable_encoder(Decimal("1.0"))
+    assert math.isclose(result, 1.0)
+
+    result = jsonable_encoder(Decimal("1"))
+    assert result == 1
+
+
+def test_openapi_encode_uuid():
+    from uuid import UUID
+
+    result = jsonable_encoder(UUID("123e4567-e89b-12d3-a456-426614174000"))
+    assert result == "123e4567-e89b-12d3-a456-426614174000"
+
+
+def test_openapi_encode_encodable():
+    from datetime import date, datetime, time
+
+    result = jsonable_encoder(date(2021, 1, 1))
+    assert result == "2021-01-01"
+
+    result = jsonable_encoder(datetime(2021, 1, 1, 0, 0, 0))
+    assert result == "2021-01-01T00:00:00"
+
+    result = jsonable_encoder(time(0, 0, 0))
+    assert result == "00:00:00"
+
+
+def test_openapi_encode_subclasses():
+    class MyColor(Color):
+        pass
+
+    result = jsonable_encoder(MyColor("red"))
+    assert result == "red"
+
+
+def test_openapi_encode_other():
+    class User:
+        def __init__(self, name: str):
+            self.name = name
+
+    result = jsonable_encoder(User(name="John"))
+    assert result == {"name": "John"}

aws_lambda_powertools/shared/types.py

@@ -0,0 +1,315 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import List
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.event_handler.api_gateway import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.models import (
+    Example,
+    Parameter,
+    ParameterInType,
+    Schema,
+)
+from aws_lambda_powertools.event_handler.openapi.params import (
+    Body,
+    Header,
+    Param,
+    ParamTypes,
+    Query,
+    _create_model_field,
+)
+from aws_lambda_powertools.shared.types import Annotated
+
+JSON_CONTENT_TYPE = "application/json"
+
+
+def test_openapi_no_params():
+    app = APIGatewayRestResolver()
+
+    @app.get("/")
+    def handler():
+        raise NotImplementedError()
+
+    schema = app.get_openapi_schema()
+    assert schema.info.title == "Powertools API"
+    assert schema.info.version == "1.0.0"
+
+    assert len(schema.paths.keys()) == 1
+    assert "/" in schema.paths
+
+    path = schema.paths["/"]
+    assert path.get
+
+    get = path.get
+    assert get.summary == "GET /"
+    assert get.operationId == "handler__get"
+
+    assert get.responses is not None
+    assert 200 in get.responses.keys()
+    response = get.responses[200]
+    assert response.description == "Successful Response"
+
+    assert JSON_CONTENT_TYPE in response.content
+    json_response = response.content[JSON_CONTENT_TYPE]
+    assert json_response.schema_ == Schema()
+    assert not json_response.examples
+    assert not json_response.encoding
+
+
+def test_openapi_with_scalar_params():
+    app = APIGatewayRestResolver()
+
+    @app.get("/users/<user_id>")
+    def handler(user_id: str, include_extra: bool = False):
+        raise NotImplementedError()
+
+    schema = app.get_openapi_schema(title="My API", version="0.2.2")
+    assert schema.info.title == "My API"
+    assert schema.info.version == "0.2.2"
+
+    assert len(schema.paths.keys()) == 1
+    assert "/users/<user_id>" in schema.paths
+
+    path = schema.paths["/users/<user_id>"]
+    assert path.get
+
+    get = path.get
+    assert get.summary == "GET /users/<user_id>"
+    assert get.operationId == "handler_users__user_id__get"
+    assert len(get.parameters) == 2
+
+    parameter = get.parameters[0]
+    assert isinstance(parameter, Parameter)
+    assert parameter.in_ == ParameterInType.path
+    assert parameter.name == "user_id"
+    assert parameter.required is True
+    assert parameter.schema_.default is None
+    assert parameter.schema_.type == "string"
+    assert parameter.schema_.title == "User Id"
+
+    parameter = get.parameters[1]
+    assert isinstance(parameter, Parameter)
+    assert parameter.in_ == ParameterInType.query
+    assert parameter.name == "include_extra"
+    assert parameter.required is False
+    assert parameter.schema_.default is False
+    assert parameter.schema_.type == "boolean"
+    assert parameter.schema_.title == "Include Extra"
+
+
+def test_openapi_with_custom_params():
+    app = APIGatewayRestResolver()
+
+    @app.get("/users", summary="Get Users", operation_id="GetUsers", description="Get paginated users", tags=["Users"])
+    def handler(
+        count: Annotated[
+            int,
+            Query(gt=0, lt=100, examples=[Example(summary="Example 1", value=10)]),
+        ] = 1,
+    ):
+        print(count)
+        raise NotImplementedError()
+
+    schema = app.get_openapi_schema()
+
+    get = schema.paths["/users"].get
+    assert len(get.parameters) == 1
+    assert get.summary == "Get Users"
+    assert get.operationId == "GetUsers"
+    assert get.description == "Get paginated users"
+    assert get.tags == ["Users"]
+
+    parameter = get.parameters[0]
+    assert parameter.required is False
+    assert parameter.name == "count"
+    assert parameter.in_ == ParameterInType.query
+    assert parameter.schema_.type == "integer"
+    assert parameter.schema_.default == 1
+    assert parameter.schema_.title == "Count"
+    assert parameter.schema_.exclusiveMinimum == 0
+    assert parameter.schema_.exclusiveMaximum == 100
+    assert len(parameter.schema_.examples) == 1
+    assert parameter.schema_.examples[0].summary == "Example 1"
+    assert parameter.schema_.examples[0].value == 10
+
+
+def test_openapi_with_scalar_returns():
+    app = APIGatewayRestResolver()
+
+    @app.get("/")
+    def handler() -> str:
+        return "Hello, world"
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    get = schema.paths["/"].get
+    assert get.parameters is None
+
+    response = get.responses[200].content[JSON_CONTENT_TYPE]
+    assert response.schema_.title == "Return"
+    assert response.schema_.type == "string"
+
+
+def test_openapi_with_pydantic_returns():
+    app = APIGatewayRestResolver()
+
+    class User(BaseModel):
+        name: str
+
+    @app.get("/")
+    def handler() -> User:
+        return User(name="Ruben Fonseca")
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    get = schema.paths["/"].get
+    assert get.parameters is None
+
+    response = get.responses[200].content[JSON_CONTENT_TYPE]
+    reference = response.schema_
+    assert reference.ref == "#/components/schemas/User"
+
+    assert "User" in schema.components.schemas
+    user_schema = schema.components.schemas["User"]
+    assert isinstance(user_schema, Schema)
+    assert user_schema.title == "User"
+    assert "name" in user_schema.properties
+
+
+def test_openapi_with_pydantic_nested_returns():
+    app = APIGatewayRestResolver()
+
+    class Order(BaseModel):
+        date: datetime
+
+    class User(BaseModel):
+        name: str
+        orders: List[Order]
+
+    @app.get("/")
+    def handler() -> User:
+        return User(name="Ruben Fonseca", orders=[Order(date=datetime.now())])
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    assert "User" in schema.components.schemas
+    assert "Order" in schema.components.schemas
+
+    user_schema = schema.components.schemas["User"]
+    assert "orders" in user_schema.properties
+    assert user_schema.properties["orders"].type == "array"
+
+
+def test_openapi_with_dataclass_return():
+    app = APIGatewayRestResolver()
+
+    @dataclass
+    class User:
+        surname: str
+
+    @app.get("/")
+    def handler() -> User:
+        return User(surname="Fonseca")
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    get = schema.paths["/"].get
+    assert get.parameters is None
+
+    response = get.responses[200].content[JSON_CONTENT_TYPE]
+    reference = response.schema_
+    assert reference.ref == "#/components/schemas/User"
+
+    assert "User" in schema.components.schemas
+    user_schema = schema.components.schemas["User"]
+    assert isinstance(user_schema, Schema)
+    assert user_schema.title == "User"
+    assert "surname" in user_schema.properties
+
+
+def test_openapi_with_body_param():
+    app = APIGatewayRestResolver()
+
+    class User(BaseModel):
+        name: str
+
+    @app.post("/users")
+    def handler(user: User):
+        print(user)
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    post = schema.paths["/users"].post
+    assert post.parameters is None
+    assert post.requestBody is not None
+
+    request_body = post.requestBody
+    assert request_body.required is True
+    assert request_body.content[JSON_CONTENT_TYPE].schema_.ref == "#/components/schemas/User"
+
+
+def test_openapi_with_embed_body_param():
+    app = APIGatewayRestResolver()
+
+    class User(BaseModel):
+        name: str
+
+    @app.post("/users")
+    def handler(user: Annotated[User, Body(embed=True)]):
+        print(user)
+
+    schema = app.get_openapi_schema()
+    assert len(schema.paths.keys()) == 1
+
+    post = schema.paths["/users"].post
+    assert post.parameters is None
+    assert post.requestBody is not None
+
+    request_body = post.requestBody
+    assert request_body.required is True
+    # Notice here we craft a specific schema for the embedded user
+    assert request_body.content[JSON_CONTENT_TYPE].schema_.ref == "#/components/schemas/Body_handler_users_post"
+
+    # Ensure that the custom body schema actually points to the real user class
+    components = schema.components
+    assert "Body_handler_users_post" in components.schemas
+    body_post_handler_schema = components.schemas["Body_handler_users_post"]
+    assert body_post_handler_schema.properties["user"].ref == "#/components/schemas/User"
+
+
+def test_create_header():
+    header = Header(convert_underscores=True)
+    assert header.convert_underscores is True
+
+
+def test_create_body():
+    body = Body(embed=True, examples=[Example(summary="Example 1", value=10)])
+    assert body.embed is True
+
+
+# Tests that when we try to create a model without a field type, we return None
+def test_create_empty_model_field():
+    result = _create_model_field(None, int, "name", False)
+    assert result is None
+
+
+# Tests that when we try to crate a param model without a source, we default to "query"
+def test_create_model_field_with_empty_in():
+    field_info = Param()
+
+    result = _create_model_field(field_info, int, "name", False)
+    assert result.field_info.in_ == ParamTypes.query
+
+
+# Tests that when we try to create a model field with convert_underscore, we convert the field name
+def test_create_model_field_convert_underscore():
+    field_info = Header(alias=None, convert_underscores=True)
+
+    result = _create_model_field(field_info, int, "user_id", False)
+    assert result.alias == "user-id"

codecov.yml

@@ -0,0 +1,49 @@
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+
+
+def test_openapi_default_response():
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/")
+    def handler():
+        pass
+
+    schema = app.get_openapi_schema()
+    responses = schema.paths["/"].get.responses
+    assert 200 in responses.keys()
+    assert responses[200].description == "Successful Response"
+
+    assert 422 in responses.keys()
+    assert responses[422].description == "Validation Error"
+
+
+def test_openapi_200_response_with_description():
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/", response_description="Custom response")
+    def handler():
+        return {"message": "hello world"}
+
+    schema = app.get_openapi_schema()
+    responses = schema.paths["/"].get.responses
+    assert 200 in responses.keys()
+    assert responses[200].description == "Custom response"
+
+    assert 422 in responses.keys()
+    assert responses[422].description == "Validation Error"
+
+
+def test_openapi_200_custom_response():
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/", responses={202: {"description": "Custom response"}})
+    def handler():
+        return {"message": "hello world"}
+
+    schema = app.get_openapi_schema()
+    responses = schema.paths["/"].get.responses
+    assert 202 in responses.keys()
+    assert responses[202].description == "Custom response"
+
+    assert 200 not in responses.keys()
+    assert 422 not in responses.keys()

pyproject.toml

@@ -0,0 +1,39 @@
+import json
+from typing import Dict
+
+import pytest
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+
+
+def test_openapi_duplicated_serialization():
+    # GIVEN APIGatewayRestResolver is initialized with enable_validation=True
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # WHEN we have duplicated operations
+    @app.get("/")
+    def handler():
+        pass
+
+    @app.get("/")
+    def handler():  # noqa: F811
+        pass
+
+    # THEN we should get a warning
+    with pytest.warns(UserWarning, match="Duplicate Operation*"):
+        app.get_openapi_schema()
+
+
+def test_openapi_serialize_json():
+    # GIVEN APIGatewayRestResolver is initialized with enable_validation=True
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/")
+    def handler():
+        pass
+
+    # WHEN we serialize as json_schema
+    schema = json.loads(app.get_openapi_json_schema())
+
+    # THEN we should get a dictionary
+    assert isinstance(schema, Dict)

ruff.toml

@@ -0,0 +1,26 @@
+from aws_lambda_powertools.event_handler.api_gateway import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.models import Server
+
+
+def test_openapi_schema_default_server():
+    app = APIGatewayRestResolver()
+
+    schema = app.get_openapi_schema(title="Hello API", version="1.0.0")
+    assert schema.servers
+    assert len(schema.servers) == 1
+    assert schema.servers[0].url == "/"
+
+
+def test_openapi_schema_custom_server():
+    app = APIGatewayRestResolver()
+
+    schema = app.get_openapi_schema(
+        title="Hello API",
+        version="1.0.0",
+        servers=[Server(url="https://example.org/", description="Example website")],
+    )
+
+    assert schema.servers
+    assert len(schema.servers) == 1
+    assert str(schema.servers[0].url) == "https://example.org/"
+    assert schema.servers[0].description == "Example website"

tests/functional/event_handler/test_openapi_encoders.py

@@ -0,0 +1,290 @@
+import json
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import PurePath
+from typing import Tuple
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.params import Body
+from aws_lambda_powertools.shared.types import Annotated
+from tests.functional.utils import load_event
+
+LOAD_GW_EVENT = load_event("apiGatewayProxyEvent.json")
+
+
+def test_validate_scalars():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # WHEN a handler is defined with a scalar parameter
+    @app.get("/users/<user_id>")
+    def handler(user_id: int):
+        print(user_id)
+
+    # sending a number
+    LOAD_GW_EVENT["path"] = "/users/123"
+
+    # THEN the handler should be invoked and return 200
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+
+    # sending a string
+    LOAD_GW_EVENT["path"] = "/users/abc"
+
+    # THEN the handler should be invoked and return 422
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 422
+    assert any(text in result["body"] for text in ["type_error.integer", "int_parsing"])
+
+
+def test_validate_scalars_with_default():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # WHEN a handler is defined with a default scalar parameter
+    @app.get("/users/<user_id>")
+    def handler(user_id: int = 123):
+        print(user_id)
+
+    # sending a number
+    LOAD_GW_EVENT["path"] = "/users/123"
+
+    # THEN the handler should be invoked and return 200
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+
+    # sending a string
+    LOAD_GW_EVENT["path"] = "/users/abc"
+
+    # THEN the handler should be invoked and return 422
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 422
+    assert any(text in result["body"] for text in ["type_error.integer", "int_parsing"])
+
+
+def test_validate_scalars_with_default_and_optional():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # WHEN a handler is defined with a default scalar parameter
+    @app.get("/users/<user_id>")
+    def handler(user_id: int = 123, include_extra: bool = False):
+        print(user_id)
+
+    # sending a number
+    LOAD_GW_EVENT["path"] = "/users/123"
+
+    # THEN the handler should be invoked and return 200
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+
+    # sending a string
+    LOAD_GW_EVENT["path"] = "/users/abc"
+
+    # THEN the handler should be invoked and return 422
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 422
+    assert any(text in result["body"] for text in ["type_error.integer", "int_parsing"])
+
+
+def test_validate_return_type():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # WHEN a handler is defined with a return type
+    @app.get("/")
+    def handler() -> int:
+        return 123
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be 123
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == 123
+
+
+def test_validate_return_tuple():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    sample_tuple = (1, 2, 3)
+
+    # WHEN a handler is defined with a return type as Tuple
+    @app.get("/")
+    def handler() -> Tuple:
+        return sample_tuple
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a tuple
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == list(sample_tuple)
+
+
+def test_validate_return_purepath():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    sample_path = PurePath(__file__)
+
+    # WHEN a handler is defined with a return type as string
+    # WHEN return value is a PurePath
+    @app.get("/")
+    def handler() -> str:
+        return sample_path
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a string
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == sample_path.as_posix()
+
+
+def test_validate_return_enum():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class Model(Enum):
+        name = "powertools"
+
+    # WHEN a handler is defined with a return type as Enum
+    @app.get("/")
+    def handler() -> Model:
+        return Model.name.value
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a string
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == "powertools"
+
+
+def test_validate_return_dataclass():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @dataclass
+    class Model:
+        name: str
+        age: int
+
+    # WHEN a handler is defined with a return type as dataclass
+    @app.get("/")
+    def handler() -> Model:
+        return Model(name="John", age=30)
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a dict
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == {"name": "John", "age": 30}
+
+
+def test_validate_return_model():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class Model(BaseModel):
+        name: str
+        age: int
+
+    # WHEN a handler is defined with a return type as Pydantic model
+    @app.get("/")
+    def handler() -> Model:
+        return Model(name="John", age=30)
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a dict
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == {"name": "John", "age": 30}
+
+
+def test_validate_invalid_return_model():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class Model(BaseModel):
+        name: str
+        age: int
+
+    # WHEN a handler is defined with a return type as Pydantic model
+    @app.get("/")
+    def handler() -> Model:
+        return {"name": "John"}  # type: ignore
+
+    LOAD_GW_EVENT["path"] = "/"
+
+    # THEN the handler should be invoked and return 422
+    # THEN the body must be a dict
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 422
+    assert "missing" in result["body"]
+
+
+def test_validate_body_param():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class Model(BaseModel):
+        name: str
+        age: int
+
+    # WHEN a handler is defined with a body parameter
+    @app.post("/")
+    def handler(user: Model) -> Model:
+        return user
+
+    LOAD_GW_EVENT["httpMethod"] = "POST"
+    LOAD_GW_EVENT["path"] = "/"
+    LOAD_GW_EVENT["body"] = json.dumps({"name": "John", "age": 30})
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a dict
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200
+    assert result["body"] == {"name": "John", "age": 30}
+
+
+def test_validate_embed_body_param():
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    class Model(BaseModel):
+        name: str
+        age: int
+
+    # WHEN a handler is defined with a body parameter
+    @app.post("/")
+    def handler(user: Annotated[Model, Body(embed=True)]) -> Model:
+        return user
+
+    LOAD_GW_EVENT["httpMethod"] = "POST"
+    LOAD_GW_EVENT["path"] = "/"
+    LOAD_GW_EVENT["body"] = json.dumps({"name": "John", "age": 30})
+
+    # THEN the handler should be invoked and return 422
+    # THEN the body must be a dict
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 422
+    assert "missing" in result["body"]
+
+    # THEN the handler should be invoked and return 200
+    # THEN the body must be a dict
+    LOAD_GW_EVENT["body"] = json.dumps({"user": {"name": "John", "age": 30}})
+    result = app(LOAD_GW_EVENT, {})
+    assert result["statusCode"] == 200