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,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Constants
+CONTENT_DISPOSITION_NAME_PARAM = "name="
+APPLICATION_JSON_CONTENT_TYPE = "application/json"
+
 
 class OpenAPIValidationMiddleware(BaseMiddlewareHandler):
     """
@@ -246,28 +251,202 @@ 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()
 
-        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
+        # If no content-type is provided, try to infer from route parameters
+        if not content_type:
+            content_type = self._infer_content_type(app)
+
+        # Handle JSON content
+        if content_type.startswith(APPLICATION_JSON_CONTENT_TYPE):
+            return self._parse_json_data(app)
+
+        # Handle URL-encoded form data
+        elif content_type.startswith("application/x-www-form-urlencoded"):
+            return self._parse_form_data(app)
+
+        # Handle multipart form data (for file uploads)
+        elif content_type.startswith("multipart/form-data"):
+            return self._parse_multipart_data(app)
+
+        else:
+            raise RequestValidationError(
+                [
+                    {
+                        "type": "content_type_invalid",
+                        "loc": ("body",),
+                        "msg": f"Unsupported content type: {content_type}",
+                        "input": {},
+                    },
+                ],
+            )
+
+    def _infer_content_type(self, app: EventHandlerInstance) -> str:
+        """Infer content type from route parameters when not explicitly provided."""
+        route = app.context.get("_route")
+        if route and route.dependant.body_params:
+            # Check if any body params are File or Form types
+            from aws_lambda_powertools.event_handler.openapi.params import File, Form
+
+            has_file_params = any(
+                isinstance(getattr(param.field_info, "__class__", None), type)
+                and issubclass(param.field_info.__class__, (File, Form))
+                for param in route.dependant.body_params
+                if hasattr(param, "field_info")
+            )
+
+            return "multipart/form-data" if has_file_params else APPLICATION_JSON_CONTENT_TYPE
+
+        # Default to JSON when no body params
+        return APPLICATION_JSON_CONTENT_TYPE
+
+    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)
+
+            # Convert list values to single values where appropriate
+            result: dict[str, Any] = {}
+            for key, values in parsed.items():
+                if len(values) == 1:
+                    result[key] = values[0]
+                else:
+                    result[key] = values  # Keep as list for multiple values
+
+            return result
+
+        except Exception as e:
+            raise RequestValidationError(
+                [
+                    {
+                        "type": "form_invalid",
+                        "loc": ("body",),
+                        "msg": "Form data parsing error",
+                        "input": {},
+                        "ctx": {"error": str(e)},
+                    },
+                ],
+            ) from e
+
+    def _parse_multipart_data(self, app: EventHandlerInstance) -> dict[str, Any]:
+        """Parse multipart form data from the request body."""
+        try:
+            content_type = app.current_event.headers.get("content-type", "")
+            body = app.current_event.decoded_body or ""
+
+            # Extract boundary from content-type header
+            boundary = self._extract_boundary(content_type)
+            if not boundary:
+                msg = "No boundary found in multipart content-type"
+                raise ValueError(msg)
+
+            # Split the body by boundary and parse each part
+            parts = body.split(f"--{boundary}")
+            result = {}
+
+            for raw_part in parts:
+                part = raw_part.strip()
+                if not part or part == "--":
+                    continue
+
+                field_name, content = self._parse_multipart_part(part)
+                if field_name:
+                    result[field_name] = content
+
+            return result
+
+        except Exception as e:
+            raise RequestValidationError(
+                [
+                    {
+                        "type": "multipart_invalid",
+                        "loc": ("body",),
+                        "msg": "Multipart data parsing error",
+                        "input": {},
+                        "ctx": {"error": str(e)},
+                    },
+                ],
+            ) from e
+
+    def _extract_boundary(self, content_type: str) -> str | None:
+        """Extract boundary from multipart content-type header."""
+        if "boundary=" in content_type:
+            return content_type.split("boundary=")[1].split(";")[0].strip()
+        return None
+
+    def _parse_multipart_part(self, part: str) -> tuple[str | None, Any]:
+        """Parse a single multipart section and return field name and content."""
+        # Split headers from content
+        if "\r\n\r\n" in part:
+            headers_section, content = part.split("\r\n\r\n", 1)
+        elif "\n\n" in part:
+            headers_section, content = part.split("\n\n", 1)
+        else:
+            return None, None
+
+        # Parse headers to find field name
+        headers = {}
+        for header_line in headers_section.split("\n"):
+            if ":" in header_line:
+                key, value = header_line.split(":", 1)
+                headers[key.strip().lower()] = value.strip()
+
+        # Extract field name from Content-Disposition header
+        content_disposition = headers.get("content-disposition", "")
+        field_name = self._extract_field_name(content_disposition)
+
+        if not field_name:
+            return None, None
+
+        # Handle file vs text field
+        if "filename=" in content_disposition:
+            # This is a file upload - convert to bytes
+            content = content.rstrip("\r\n")
+            return field_name, content.encode() if isinstance(content, str) else content
         else:
-            raise NotImplementedError("Only JSON body is supported")
+            # This is a text field - keep as string
+            return field_name, content.rstrip("\r\n")
+
+    def _extract_field_name(self, content_disposition: str) -> str | None:
+        """Extract field name from Content-Disposition header."""
+        if CONTENT_DISPOSITION_NAME_PARAM not in content_disposition:
+            return None
+
+        # Handle both quoted and unquoted names
+        if 'name="' in content_disposition:
+            name_start = content_disposition.find('name="') + 6
+            name_end = content_disposition.find('"', name_start)
+            return content_disposition[name_start:name_end]
+        elif CONTENT_DISPOSITION_NAME_PARAM in content_disposition:
+            name_start = content_disposition.find(CONTENT_DISPOSITION_NAME_PARAM) + len(CONTENT_DISPOSITION_NAME_PARAM)
+            name_end = content_disposition.find(";", name_start)
+            if name_end == -1:
+                name_end = len(content_disposition)
+            return content_disposition[name_start:name_end].strip()
+
+        return None
 
 
 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,
+    File,
+    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
 
 
@@ -366,12 +367,12 @@ def get_body_field_info(
     if not required:
         body_field_info_kwargs["default"] = None
 
-    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")
+    if any(isinstance(f.field_info, File) for f in flat_dependant.body_params):
+        body_field_info = Body
+        body_field_info_kwargs["media_type"] = "multipart/form-data"
+    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,13 @@ def __init__(
         json_schema_extra: dict[str, Any] | None = None,
         **extra: Any,
     ):
+        # For file uploads, ensure the OpenAPI schema has the correct format
+        file_schema_extra = {"format": "binary"}
+        if json_schema_extra:
+            json_schema_extra.update(file_schema_extra)
+        else:
+            json_schema_extra = file_schema_extra
+
         super().__init__(
             default=default,
             default_factory=default_factory,

docs/core/event_handler/api_gateway.md

@@ -523,6 +523,47 @@ 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 file uploads and form data
+
+!!! info "You must set `enable_validation=True` to handle file uploads and form data via type annotation."
+
+We use the `Annotated` type to tell the Event Handler that a parameter expects file upload or form data. This automatically sets the correct OpenAPI schema for `multipart/form-data` requests.
+
+In the following example, we use `File` and `Form` OpenAPI types to handle file uploads and form fields:
+
+* `File` parameters expect binary file data and generate OpenAPI schema with `format: binary`
+* `Form` parameters expect form field values from multipart form data
+* The OpenAPI spec will automatically set `requestBody` content type to `multipart/form-data`
+
+=== "handling_file_uploads.py"
+
+    ```python hl_lines="5 9-10 18-19"
+    --8<-- "examples/event_handler_rest/src/handling_file_uploads.py"
+    ```
+
+    1. If you're not using Python 3.9 or higher, you can install and use [`typing_extensions`](https://pypi.org/project/typing-extensions/){target="_blank" rel="nofollow"} to the same effect
+    2. `File` is a special OpenAPI type for binary file uploads that sets `format: binary` in the schema
+    3. `Form` is a special OpenAPI type for form field values in multipart requests
+
+=== "Multiple files"
+
+    You can handle multiple file uploads by declaring parameters as lists:
+
+    ```python hl_lines="9-10"
+    --8<-- "examples/event_handler_rest/src/handling_multiple_file_uploads.py"
+    ```
+
+    1. `files` will be a list containing the binary data of each uploaded file
+
+???+ note "OpenAPI Schema Generation"
+    When you use `File` or `Form` parameters, the generated OpenAPI specification will automatically include:
+
+    * `requestBody` with content type `multipart/form-data`
+    * Proper schema definitions with `format: binary` for file parameters
+    * Form field descriptions and constraints
+
+    This ensures API documentation tools like SwaggerUI correctly display file upload interfaces.
+
 #### Supported types for response serialization
 
 With data validation enabled, we natively support serializing the following data types to JSON:

examples/event_handler_rest/src/handling_file_uploads.py

@@ -0,0 +1,20 @@
+from typing import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.params import File, Form
+
+app = APIGatewayRestResolver(enable_validation=True)
+
+
+@app.post("/upload")
+def upload_file(
+    file: Annotated[bytes, File(description="File to upload")],
+    filename: Annotated[str, Form(description="Name of the file")],
+):
+    # file contains the binary data of the uploaded file
+    # filename contains the form field value
+    return {"message": f"Uploaded {filename}", "size": len(file)}
+
+
+def lambda_handler(event, context):
+    return app.resolve(event, context)