feat(event_handler): add support for form data in OpenAPI utility

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -5,6 +5,7 @@
 import logging
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence
+from urllib.parse import parse_qs
 
 from pydantic import BaseModel
 
@@ -30,6 +31,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Constants
+CONTENT_DISPOSITION_NAME_PARAM = "name="
+APPLICATION_JSON_CONTENT_TYPE = "application/json"
+APPLICATION_FORM_CONTENT_TYPE = "application/x-www-form-urlencoded"
+
 
 class OpenAPIValidationMiddleware(BaseMiddlewareHandler):
     """
@@ -246,28 +252,61 @@ def _prepare_response_content(
 
     def _get_body(self, app: EventHandlerInstance) -> dict[str, Any]:
         """
-        Get the request body from the event, and parse it as JSON.
+        Get the request body from the event, and parse it according to content type.
         """
+        content_type = app.current_event.headers.get("content-type", "").strip()
+
+        # Handle JSON content
+        if not content_type or content_type.startswith(APPLICATION_JSON_CONTENT_TYPE):
+            return self._parse_json_data(app)
+
+        # Handle URL-encoded form data
+        elif content_type.startswith(APPLICATION_FORM_CONTENT_TYPE):
+            return self._parse_form_data(app)
 
-        content_type = app.current_event.headers.get("content-type")
-        if not content_type or content_type.strip().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")
+            raise NotImplementedError("Only JSON body or Form() are supported")
+
+    def _parse_json_data(self, app: EventHandlerInstance) -> dict[str, Any]:
+        """Parse JSON data from the request body."""
+        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
+
+    def _parse_form_data(self, app: EventHandlerInstance) -> dict[str, Any]:
+        """Parse URL-encoded form data from the request body."""
+        try:
+            body = app.current_event.decoded_body or ""
+            # parse_qs returns dict[str, list[str]], but we want dict[str, str] for single values
+            parsed = parse_qs(body, keep_blank_values=True)
+
+            result: dict[str, Any] = {key: values[0] if len(values) == 1 else values for key, values in parsed.items()}
+            return result
+
+        except Exception as e:  # pragma: no cover
+            raise RequestValidationError(  # pragma: no cover
+                [
+                    {
+                        "type": "form_invalid",
+                        "loc": ("body",),
+                        "msg": "Form data parsing error",
+                        "input": {},
+                        "ctx": {"error": str(e)},
+                    },
+                ],
+            ) from e
 
 
 def _request_params_to_args(

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -14,12 +14,12 @@
 from aws_lambda_powertools.event_handler.openapi.params import (
     Body,
     Dependant,
+    Form,
     Header,
     Param,
     ParamTypes,
     Query,
     _File,
-    _Form,
     analyze_param,
     create_response_field,
     get_flat_dependant,
@@ -348,6 +348,7 @@ def get_body_field(*, dependant: Dependant, name: str) -> ModelField | None:
         alias="body",
         field_info=body_field_info(**body_field_info_kwargs),
     )
+
     return final_field
 
 
@@ -369,9 +370,9 @@ def get_body_field_info(
     if any(isinstance(f.field_info, _File) for f in flat_dependant.body_params):
         # MAINTENANCE: body_field_info: type[Body] = _File
         raise NotImplementedError("_File fields are not supported in request bodies")
-    elif any(isinstance(f.field_info, _Form) for f in flat_dependant.body_params):
-        # MAINTENANCE: body_field_info: type[Body] = _Form
-        raise NotImplementedError("_Form fields are not supported in request bodies")
+    elif any(isinstance(f.field_info, Form) for f in flat_dependant.body_params):
+        body_field_info = Body
+        body_field_info_kwargs["media_type"] = "application/x-www-form-urlencoded"
     else:
         body_field_info = Body
 

aws_lambda_powertools/event_handler/openapi/params.py

@@ -737,9 +737,9 @@ def __repr__(self) -> str:
         return f"{self.__class__.__name__}({self.default})"
 
 
-class _Form(Body):
+class Form(Body):
     """
-    A class used internally to represent a form parameter in a path operation.
+    A class used to represent a form parameter in a path operation.
     """
 
     def __init__(
@@ -809,9 +809,9 @@ def __init__(
         )
 
 
-class _File(_Form):
+class _File(Form):
     """
-    A class used internally to represent a file parameter in a path operation.
+    A class used to represent a file parameter in a path operation.
     """
 
     def __init__(
@@ -848,6 +848,14 @@ def __init__(
         json_schema_extra: dict[str, Any] | None = None,
         **extra: Any,
     ):
+        # For file uploads, ensure the OpenAPI schema has the correct format
+        # Also we can't test it
+        file_schema_extra = {"format": "binary"}  # pragma: no cover
+        if json_schema_extra:  # pragma: no cover
+            json_schema_extra.update(file_schema_extra)  # pragma: no cover
+        else:  # pragma: no cover
+            json_schema_extra = file_schema_extra  # pragma: no cover
+
         super().__init__(
             default=default,
             default_factory=default_factory,

docs/core/event_handler/api_gateway.md

@@ -523,6 +523,18 @@ In the following example, we use a new `Header` OpenAPI type to add [one out of
 
     1. `cloudfront_viewer_country` is a list that must contain values from the `CountriesAllowed` enumeration.
 
+#### Handling form data
+
+!!! info "You must set `enable_validation=True` to handle file uploads and form data via type annotation."
+
+You can use the `Form` type to tell the Event Handler that a parameter expects file upload or form data. This automatically sets the correct OpenAPI schema for `application/x-www-form-urlencoded` requests.
+
+=== "working_with_form_data.py"
+
+    ```python hl_lines="4 11 12"
+    --8<-- "examples/event_handler_rest/src/working_with_form_data.py"
+    ```
+
 #### Supported types for response serialization
 
 With data validation enabled, we natively support serializing the following data types to JSON:

examples/event_handler_rest/src/working_with_form_data.py

@@ -0,0 +1,19 @@
+from typing import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.params import Form
+
+app = APIGatewayRestResolver(enable_validation=True)
+
+
+@app.post("/submit_form")
+def upload_file(
+    name: Annotated[str, Form(description="Your name")],
+    age: Annotated[str, Form(description="Your age")],
+):
+    # You can access form data
+    return {"message": f"Your name is {name} and age is {age}"}
+
+
+def lambda_handler(event, context):
+    return app.resolve(event, context)

tests/functional/event_handler/_pydantic/test_openapi_params.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 
 from pydantic import BaseModel, Field
 from typing_extensions import Annotated
@@ -14,6 +14,7 @@
 )
 from aws_lambda_powertools.event_handler.openapi.params import (
     Body,
+    Form,
     Header,
     Param,
     ParamTypes,
@@ -649,3 +650,129 @@ def handler(
     assert parameter.schema_.type == "integer"
     assert parameter.schema_.default == 1
     assert parameter.schema_.title == "Count"
+
+
+def test_openapi_form_only_parameters():
+    """Test Form parameters generate application/x-www-form-urlencoded content type."""
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/form-data")
+    def create_form_data(
+        name: Annotated[str, Form(description="User name")],
+        email: Annotated[str, Form(description="User email")] = "test@example.com",
+    ):
+        return {"name": name, "email": email}
+
+    schema = app.get_openapi_schema()
+
+    # Check that the endpoint is present
+    assert "/form-data" in schema.paths
+
+    post_op = schema.paths["/form-data"].post
+    assert post_op is not None
+
+    # Check request body
+    request_body = post_op.requestBody
+    assert request_body is not None
+
+    # Check content type is application/x-www-form-urlencoded
+    assert "application/x-www-form-urlencoded" in request_body.content
+
+    # Get the schema reference
+    form_content = request_body.content["application/x-www-form-urlencoded"]
+    assert form_content.schema_ is not None
+
+    # Check that it references a component schema
+    schema_ref = form_content.schema_.ref
+    assert schema_ref is not None
+    assert schema_ref.startswith("#/components/schemas/")
+
+    # Get the component schema
+    component_name = schema_ref.split("/")[-1]
+    assert component_name in schema.components.schemas
+
+    component_schema = schema.components.schemas[component_name]
+    properties = component_schema.properties
+
+    # Check form parameters
+    assert "name" in properties
+    name_prop = properties["name"]
+    assert name_prop.type == "string"
+    assert name_prop.description == "User name"
+
+    assert "email" in properties
+    email_prop = properties["email"]
+    assert email_prop.type == "string"
+    assert email_prop.description == "User email"
+    assert email_prop.default == "test@example.com"
+
+    # Check required fields (only name should be required since email has default)
+    assert component_schema.required == ["name"]
+
+
+def test_openapi_mixed_body_media_types():
+    """Test mixed Body parameters with different media types."""
+
+    class UserData(BaseModel):
+        name: str
+        email: str
+
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/mixed-body")
+    def mixed_body_endpoint(user_data: Annotated[UserData, Body(media_type="application/json")]):
+        return {"status": "created"}
+
+    schema = app.get_openapi_schema()
+
+    # Check that the endpoint uses the specified media type
+    assert "/mixed-body" in schema.paths
+
+    post_op = schema.paths["/mixed-body"].post
+    request_body = post_op.requestBody
+
+    # Should use the specified media type
+    assert "application/json" in request_body.content
+
+
+def test_openapi_form_parameter_edge_cases():
+    """Test Form parameters with various edge cases."""
+
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/form-edge-cases")
+    def form_edge_cases(
+        required_field: Annotated[str, Form(description="Required field")],
+        optional_field: Annotated[Optional[str], Form(description="Optional field")] = None,
+        field_with_default: Annotated[str, Form(description="Field with default")] = "default_value",
+    ):
+        return {"required": required_field, "optional": optional_field, "default": field_with_default}
+
+    schema = app.get_openapi_schema()
+
+    # Check that the endpoint is present
+    assert "/form-edge-cases" in schema.paths
+
+    post_op = schema.paths["/form-edge-cases"].post
+    request_body = post_op.requestBody
+
+    # Should use application/x-www-form-urlencoded for form-only parameters
+    assert "application/x-www-form-urlencoded" in request_body.content
+
+    # Get the component schema
+    form_content = request_body.content["application/x-www-form-urlencoded"]
+    schema_ref = form_content.schema_.ref
+    component_name = schema_ref.split("/")[-1]
+    component_schema = schema.components.schemas[component_name]
+
+    properties = component_schema.properties
+
+    # Check all fields are present
+    assert "required_field" in properties
+    assert "optional_field" in properties
+    assert "field_with_default" in properties
+
+    # Check required vs optional handling
+    assert "required_field" in component_schema.required
+    assert "optional_field" not in component_schema.required  # Optional
+    assert "field_with_default" not in component_schema.required  # Has default