feat(event-handler): Add UploadFile class for file metadata access

- Add FastAPI-inspired UploadFile class with filename, content_type, size, headers properties
- Enhance multipart parser to extract and preserve file metadata from Content-Disposition headers
- Implement automatic type resolution for backward compatibility with existing bytes-based File parameters
- Add comprehensive Pydantic schema validation for UploadFile class
- Include 6 comprehensive test cases covering metadata access, backward compatibility, and file reconstruction scenarios
- Update official example to showcase both new UploadFile and legacy bytes approaches
- Maintain 100% backward compatibility - existing bytes code works unchanged
- Address @leandrodamascena feedback about file reconstruction capabilities in Lambda environments

Fixes: File parameter enhancement for metadata access in AWS Lambda file uploads

Author: oyiz-michael

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -5,7 +5,7 @@
 import logging
 import re
 from copy import deepcopy
-from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence
+from typing import TYPE_CHECKING, Any, Callable, Mapping, MutableMapping, Sequence, Union
 from urllib.parse import parse_qs
 
 from pydantic import BaseModel
@@ -20,7 +20,7 @@
 from aws_lambda_powertools.event_handler.openapi.dependant import is_scalar_field
 from aws_lambda_powertools.event_handler.openapi.encoders import jsonable_encoder
 from aws_lambda_powertools.event_handler.openapi.exceptions import RequestValidationError, ResponseValidationError
-from aws_lambda_powertools.event_handler.openapi.params import Param
+from aws_lambda_powertools.event_handler.openapi.params import Param, UploadFile
 
 if TYPE_CHECKING:
     from aws_lambda_powertools.event_handler import Response
@@ -245,7 +245,7 @@ def _parse_multipart_sections(self, decoded_bytes: bytes, boundary_bytes: bytes)
 
         return parsed_data
 
-    def _parse_multipart_section(self, section: bytes) -> tuple[str | None, bytes | str]:
+    def _parse_multipart_section(self, section: bytes) -> tuple[str | None, bytes | str | UploadFile]:
         """Parse a single multipart section to extract field name and content."""
         headers_part, content = self._split_section_headers_and_content(section)
 
@@ -261,8 +261,30 @@ def _parse_multipart_section(self, section: bytes) -> tuple[str | None, bytes |
 
         # Check if it's a file field and process accordingly
         if "filename=" in headers_part:
-            # It's a file - store as bytes
-            return field_name, content
+            # It's a file - extract metadata and create UploadFile
+            filename_match = re.search(r'filename="([^"]*)"', headers_part)
+            filename = filename_match.group(1) if filename_match else None
+
+            # Extract Content-Type if present
+            content_type_match = re.search(r"Content-Type:\s*([^\r\n]+)", headers_part, re.IGNORECASE)
+            content_type = content_type_match.group(1).strip() if content_type_match else None
+
+            # Parse all headers from the section
+            headers = {}
+            for line_raw in headers_part.split("\n"):
+                line = line_raw.strip()
+                if ":" in line and not line.startswith("Content-Disposition"):
+                    key, value = line.split(":", 1)
+                    headers[key.strip()] = value.strip()
+
+            # Create UploadFile instance with metadata
+            upload_file = UploadFile(
+                file=content,
+                filename=filename,
+                content_type=content_type,
+                headers=headers,
+            )
+            return field_name, upload_file
         else:
             # It's a regular form field - decode as string
             return field_name, self._decode_form_field_content(content)
@@ -509,6 +531,27 @@ def _request_body_to_args(
             continue
 
         # MAINTENANCE: Handle byte and file fields
+        # Check if we have an UploadFile but the field expects bytes
+        from typing import get_args, get_origin
+
+        field_type = field.type_
+
+        # Handle Union types (e.g., Union[bytes, None] for optional parameters)
+        if get_origin(field_type) is Union:
+            # Get the non-None types from the Union
+            union_args = get_args(field_type)
+            non_none_types = [arg for arg in union_args if arg is not type(None)]
+            if non_none_types:
+                field_type = non_none_types[0]  # Use the first non-None type
+
+        if isinstance(value, UploadFile) and field_type is bytes:
+            # Convert UploadFile to bytes for backward compatibility
+            value = value.file
+        elif isinstance(value, bytes) and field_type == UploadFile:
+            # Convert bytes to UploadFile if that's what's expected
+            # This shouldn't normally happen in our current implementation,
+            # but provides a fallback path
+            value = UploadFile(file=value)
 
         # Finally, validate the value
         values[field.name] = _validate_field(field=field, value=value, loc=loc, existing_errors=errors)

aws_lambda_powertools/event_handler/openapi/params.py

@@ -29,7 +29,105 @@
 This turns the low-level function signature into typed, validated Pydantic models for consumption.
 """
 
-__all__ = ["Path", "Query", "Header", "Body", "Form", "File"]
+__all__ = ["Path", "Query", "Header", "Body", "Form", "File", "UploadFile"]
+
+
+class UploadFile:
+    """
+    A file uploaded as part of a multipart/form-data request.
+
+    Similar to FastAPI's UploadFile, this class provides access to both file content
+    and metadata such as filename, content type, and headers.
+
+    Example:
+        ```python
+        @app.post("/upload")
+        def upload_file(file: Annotated[UploadFile, File()]):
+            return {
+                "filename": file.filename,
+                "content_type": file.content_type,
+                "size": file.size,
+                "content": file.file.decode() if file.size < 1000 else "File too large to display"
+            }
+        ```
+    """
+
+    def __init__(
+        self,
+        file: bytes,
+        filename: str | None = None,
+        content_type: str | None = None,
+        headers: dict[str, str] | None = None,
+    ):
+        """
+        Initialize an UploadFile instance.
+
+        Parameters
+        ----------
+        file : bytes
+            The file content as bytes
+        filename : str | None
+            The original filename from the Content-Disposition header
+        content_type : str | None
+            The content type from the Content-Type header
+        headers : dict[str, str] | None
+            All headers from the multipart section
+        """
+        self.file = file
+        self.filename = filename
+        self.content_type = content_type
+        self.headers = headers or {}
+
+    @property
+    def size(self) -> int:
+        """Return the size of the file in bytes."""
+        return len(self.file)
+
+    def read(self, size: int = -1) -> bytes:
+        """
+        Read and return up to size bytes from the file.
+
+        Parameters
+        ----------
+        size : int
+            Number of bytes to read. If -1 (default), read the entire file.
+
+        Returns
+        -------
+        bytes
+            The file content
+        """
+        if size == -1:
+            return self.file
+        return self.file[:size]
+
+    def __repr__(self) -> str:
+        """Return a string representation of the UploadFile."""
+        return f"UploadFile(filename={self.filename!r}, size={self.size}, content_type={self.content_type!r})"
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        _source_type: Any,
+        _handler: Any,
+    ) -> Any:
+        """Return Pydantic core schema for UploadFile."""
+        from pydantic_core import core_schema
+
+        # Define the schema for UploadFile validation
+        return core_schema.no_info_plain_validator_function(
+            cls._validate,
+            serialization=core_schema.to_string_ser_schema(),
+        )
+
+    @classmethod
+    def _validate(cls, value: Any) -> UploadFile:
+        """Validate and convert value to UploadFile."""
+        if isinstance(value, cls):
+            return value
+        if isinstance(value, bytes):
+            return cls(file=value)
+        raise ValueError(f"Expected UploadFile or bytes, got {type(value)}")
 
 
 class ParamTypes(Enum):

examples/event_handler_rest/src/file_parameter_example.py

@@ -1,31 +1,77 @@
 """
 Example demonstrating File parameter usage for handling file uploads.
+This showcases both the new UploadFile class for metadata access and
+backward-compatible bytes approach.
 """
 
 from __future__ import annotations
 
 from typing import Annotated, Union
 
 from aws_lambda_powertools.event_handler import APIGatewayRestResolver
-from aws_lambda_powertools.event_handler.openapi.params import File, Form
+from aws_lambda_powertools.event_handler.openapi.params import File, Form, UploadFile
 
 # Initialize resolver with OpenAPI validation enabled
 app = APIGatewayRestResolver(enable_validation=True)
 
 
+# ========================================
+# NEW: UploadFile with Metadata Access
+# ========================================
+
+
+@app.post("/upload-with-metadata")
+def upload_file_with_metadata(file: Annotated[UploadFile, File(description="File with metadata access")]):
+    """Upload a file with full metadata access - NEW UploadFile feature!"""
+    return {
+        "status": "uploaded",
+        "filename": file.filename,
+        "content_type": file.content_type,
+        "file_size": file.size,
+        "headers": file.headers,
+        "content_preview": file.read(100).decode("utf-8", errors="ignore"),
+        "can_reconstruct_file": True,
+        "message": "File uploaded with metadata access",
+    }
+
+
+@app.post("/upload-mixed-form")
+def upload_file_with_form_data(
+    file: Annotated[UploadFile, File(description="File with metadata")],
+    description: Annotated[str, Form(description="File description")],
+    category: Annotated[str | None, Form(description="File category")] = None,
+):
+    """Upload file with UploadFile metadata + form data."""
+    return {
+        "status": "uploaded",
+        "filename": file.filename,
+        "content_type": file.content_type,
+        "file_size": file.size,
+        "description": description,
+        "category": category,
+        "custom_headers": {k: v for k, v in file.headers.items() if k.startswith("X-")},
+        "message": "File and form data uploaded with metadata",
+    }
+
+
+# ========================================
+# BACKWARD COMPATIBLE: Bytes Approach
+# ========================================
+
+
 @app.post("/upload")
 def upload_single_file(file: Annotated[bytes, File(description="File to upload")]):
-    """Upload a single file."""
+    """Upload a single file - LEGACY bytes approach (still works!)."""
     return {"status": "uploaded", "file_size": len(file), "message": "File uploaded successfully"}
 
 
-@app.post("/upload-with-metadata")
-def upload_file_with_metadata(
+@app.post("/upload-legacy-metadata")
+def upload_file_legacy_with_metadata(
     file: Annotated[bytes, File(description="File to upload")],
     description: Annotated[str, Form(description="File description")],
     tags: Annotated[Union[str, None], Form(description="Optional tags")] = None,  # noqa: UP007
 ):
-    """Upload a file with additional form metadata."""
+    """Upload a file with additional form metadata - LEGACY bytes approach."""
     return {
         "status": "uploaded",
         "file_size": len(file),
@@ -37,22 +83,24 @@ def upload_file_with_metadata(
 
 @app.post("/upload-multiple")
 def upload_multiple_files(
-    primary_file: Annotated[bytes, File(alias="primary", description="Primary file")],
-    secondary_file: Annotated[bytes, File(alias="secondary", description="Secondary file")],
+    primary_file: Annotated[UploadFile, File(alias="primary", description="Primary file with metadata")],
+    secondary_file: Annotated[bytes, File(alias="secondary", description="Secondary file as bytes")],
 ):
-    """Upload multiple files."""
+    """Upload multiple files - showcasing BOTH UploadFile and bytes approaches."""
     return {
         "status": "uploaded",
-        "primary_size": len(primary_file),
+        "primary_filename": primary_file.filename,
+        "primary_content_type": primary_file.content_type,
+        "primary_size": primary_file.size,
         "secondary_size": len(secondary_file),
-        "total_size": len(primary_file) + len(secondary_file),
-        "message": "Multiple files uploaded successfully",
+        "total_size": primary_file.size + len(secondary_file),
+        "message": "Multiple files uploaded with mixed approaches",
     }
 
 
 @app.post("/upload-with-constraints")
 def upload_small_file(file: Annotated[bytes, File(description="Small file only", max_length=1024)]):
-    """Upload a file with size constraints (max 1KB)."""
+    """Upload a file with size constraints (max 1KB) - bytes approach."""
     return {
         "status": "uploaded",
         "file_size": len(file),
@@ -63,14 +111,16 @@ def upload_small_file(file: Annotated[bytes, File(description="Small file only",
 @app.post("/upload-optional")
 def upload_optional_file(
     message: Annotated[str, Form(description="Required message")],
-    file: Annotated[Union[bytes, None], File(description="Optional file")] = None,  # noqa: UP007
+    file: Annotated[UploadFile | None, File(description="Optional file with metadata")] = None,
 ):
-    """Upload with an optional file parameter."""
+    """Upload with an optional UploadFile parameter - NEW approach!"""
     return {
         "status": "processed",
         "message": message,
         "has_file": file is not None,
-        "file_size": len(file) if file else 0,
+        "filename": file.filename if file else None,
+        "content_type": file.content_type if file else None,
+        "file_size": file.size if file else 0,
     }
 
 
@@ -80,13 +130,28 @@ def lambda_handler(event, context):
     return app.resolve(event, context)
 
 
-# The File parameter provides:
-# 1. Automatic multipart/form-data parsing
-# 2. OpenAPI schema generation with proper file upload documentation
-# 3. Request validation with meaningful error messages
-# 4. Support for file constraints (max_length, etc.)
-# 5. Compatibility with WebKit and other browser boundary formats
-# 6. Base64-encoded request handling (common in AWS Lambda)
-# 7. Mixed file and form data support
-# 8. Multiple file upload support
-# 9. Optional file parameters
+# The File parameter now provides TWO approaches:
+#
+# 1. NEW UploadFile Class (Recommended):
+#    - filename property (e.g., "document.pdf")
+#    - content_type property (e.g., "application/pdf")
+#    - size property (file size in bytes)
+#    - headers property (dict of all multipart headers)
+#    - read() method (flexible content access)
+#    - Perfect for file reconstruction in Lambda/S3 scenarios
+#
+# 2. LEGACY bytes approach (Backward Compatible):
+#    - Direct bytes content access
+#    - Existing code continues to work unchanged
+#    - Automatic conversion from UploadFile to bytes when needed
+#
+# Both approaches provide:
+# - Automatic multipart/form-data parsing
+# - OpenAPI schema generation with proper file upload documentation
+# - Request validation with meaningful error messages
+# - Support for file constraints (max_length, etc.)
+# - Compatibility with WebKit and other browser boundary formats
+# - Base64-encoded request handling (common in AWS Lambda)
+# - Mixed file and form data support
+# - Multiple file upload support
+# - Optional file parameters