fix: add OpenAPI schema support for UploadFile class

Author: oyiz-michael

aws_lambda_powertools/event_handler/openapi/params.py

@@ -115,10 +115,19 @@ def __get_pydantic_core_schema__(
         from pydantic_core import core_schema
 
         # Define the schema for UploadFile validation
-        return core_schema.no_info_plain_validator_function(
+        schema = core_schema.no_info_plain_validator_function(
             cls._validate,
             serialization=core_schema.to_string_ser_schema(),
         )
+        
+        # Add OpenAPI schema info
+        schema["json_schema_extra"] = {
+            "type": "string",
+            "format": "binary",
+            "description": "A file uploaded as part of a multipart/form-data request",
+        }
+        
+        return schema
 
     @classmethod
     def _validate(cls, value: Any) -> UploadFile:
@@ -128,6 +137,28 @@ def _validate(cls, value: Any) -> UploadFile:
         if isinstance(value, bytes):
             return cls(file=value)
         raise ValueError(f"Expected UploadFile or bytes, got {type(value)}")
+        
+    @classmethod
+    def __get_validators__(cls):
+        """Return validators for Pydantic v1 compatibility."""
+        yield cls._validate
+        
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, _core_schema: Any, field_schema: Any
+    ) -> dict[str, Any]:
+        """Modify the JSON schema for OpenAPI compatibility."""
+        # Handle both Pydantic v1 and v2 schemas
+        json_schema = field_schema(_core_schema) if callable(field_schema) else {}
+        
+        # Add binary file format for OpenAPI
+        json_schema.update(
+            type="string",
+            format="binary",
+            description="A file uploaded as part of a multipart/form-data request",
+        )
+        
+        return json_schema
 
 
 class ParamTypes(Enum):

examples/event_handler/upload_file_example.py

@@ -0,0 +1,79 @@
+"""
+Example of using UploadFile with OpenAPI schema generation
+
+This example demonstrates how to use the UploadFile class with FastAPI-like
+file handling and proper OpenAPI schema generation.
+"""
+
+from typing_extensions import Annotated, List
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.params import File, Form, UploadFile
+
+app = APIGatewayRestResolver()
+
+
+@app.post("/upload")
+def upload_file(file: Annotated[UploadFile, File()]):
+    """
+    Upload a single file.
+    
+    Returns file metadata and a preview of the content.
+    """
+    return {
+        "filename": file.filename,
+        "content_type": file.content_type,
+        "size": file.size,
+        "content_preview": file.file[:100].decode() if file.size < 10000 else "Content too large to preview",
+    }
+
+
+@app.post("/upload-multiple")
+def upload_multiple_files(
+    primary_file: Annotated[UploadFile, File(alias="primary", description="Primary file with metadata")],
+    secondary_file: Annotated[bytes, File(alias="secondary", description="Secondary file as bytes")],
+    description: Annotated[str, Form(description="Description of the uploaded files")],
+):
+    """
+    Upload multiple files with form data.
+    
+    Shows how to mix UploadFile, bytes files, and form data in the same endpoint.
+    """
+    return {
+        "status": "uploaded",
+        "description": description,
+        "primary_filename": primary_file.filename,
+        "primary_content_type": primary_file.content_type,
+        "primary_size": primary_file.size,
+        "secondary_size": len(secondary_file),
+        "total_size": primary_file.size + len(secondary_file),
+    }
+
+
+@app.post("/upload-with-headers")
+def upload_with_headers(file: Annotated[UploadFile, File()]):
+    """
+    Upload a file and access its headers.
+    
+    Demonstrates how to access all headers from the multipart section.
+    """
+    return {
+        "filename": file.filename,
+        "content_type": file.content_type,
+        "size": file.size,
+        "headers": file.headers,
+    }
+
+
+def handler(event, context):
+    return app.resolve(event, context)
+
+
+if __name__ == "__main__":
+    # Print the OpenAPI schema for testing
+    schema = app.get_openapi_schema(title="File Upload API", version="1.0.0")
+    print("\n✅ OpenAPI schema generated successfully!")
+    
+    # You can access the schema as JSON with:
+    # import json
+    # print(json.dumps(schema.model_dump(), indent=2))

tests/functional/event_handler/_pydantic/test_uploadfile_openapi_schema.py

@@ -0,0 +1,72 @@
+import pytest
+import json
+from typing_extensions import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+
+class TestUploadFileOpenAPISchema:
+    """Test UploadFile OpenAPI schema generation."""
+
+    def test_upload_file_openapi_schema(self):
+        """Test OpenAPI schema generation with UploadFile."""
+        app = APIGatewayRestResolver()
+
+        @app.post("/upload-single")
+        def upload_single_file(file: Annotated[UploadFile, File()]):
+            """Upload a single file."""
+            return {"filename": file.filename, "size": file.size}
+
+        @app.post("/upload-multiple")
+        def upload_multiple_files(
+            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 - showcasing BOTH UploadFile and bytes approaches."""
+            return {
+                "status": "uploaded",
+                "primary_filename": primary_file.filename,
+                "primary_content_type": primary_file.content_type,
+                "primary_size": primary_file.size,
+                "secondary_size": len(secondary_file),
+                "total_size": primary_file.size + len(secondary_file),
+            }
+
+        # Generate OpenAPI schema
+        schema = app.get_openapi_schema()
+        
+        # Print schema for debugging
+        schema_dict = schema.model_dump()
+        print("SCHEMA PATHS:")
+        for path, path_item in schema_dict["paths"].items():
+            print(f"Path: {path}")
+            if "post" in path_item:
+                if "requestBody" in path_item["post"]:
+                    if "content" in path_item["post"]["requestBody"]:
+                        if "multipart/form-data" in path_item["post"]["requestBody"]["content"]:
+                            print("  Found multipart/form-data")
+                            print(f"  Schema: {json.dumps(path_item['post']['requestBody']['content']['multipart/form-data'], indent=2)}")
+        
+        print("\nSCHEMA COMPONENTS:")
+        if "components" in schema_dict and "schemas" in schema_dict["components"]:
+            for name, comp_schema in schema_dict["components"]["schemas"].items():
+                if "file" in name.lower() or "upload" in name.lower():
+                    print(f"Component: {name}")
+                    print(f"  {json.dumps(comp_schema, indent=2)}")
+        
+        # Basic verification
+        paths = schema.paths
+        assert "/upload-single" in paths
+        assert "/upload-multiple" in paths
+        
+        # Verify upload-single endpoint exists
+        upload_single = paths["/upload-single"]
+        assert upload_single.post is not None
+        
+        # Verify upload-multiple endpoint exists
+        upload_multiple = paths["/upload-multiple"]
+        assert upload_multiple.post is not None
+        
+        # Print success
+        print("\n✅ Basic OpenAPI schema generation tests passed")