How to Handle Form Data in AWS Lambda APIs with Powertools OpenAPI Support

#aws #lambda #openai #serverless

A complete guide to using the new Form parameter support in AWS Lambda Powertools for Python

AWS Lambda Powertools for Python now supports form data parameters in OpenAPI schema generation! This means you can build Lambda APIs that accept application/x-www-form-urlencoded data with automatic validation and documentation.

Why This Matters
Before this feature, Lambda APIs built with Powertools could only generate proper OpenAPI schemas for JSON payloads. If you needed to handle form data (like HTML forms or certain client applications), you had to:

Manually parse form data from the raw request body
Write custom validation logic
Maintain separate API documentation
Handle errors without proper validation feedback

Now you can handle form data declaratively with automatic validation, error handling, and OpenAPI documentation generation.

Getting Started

Installation
Make sure you have the latest version of AWS Lambda Powertools:
pip install aws-lambda-powertools[validation]

Basic Form Handling
Here's how to create a Lambda function that accepts form data:

from typing import Annotated
from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.openapi.params import Form

logger = Logger()
app = APIGatewayRestResolver(enable_validation=True)

@app.post("/contact")
def submit_contact_form(
    name: Annotated[str, Form(description="Contact's full name")],
    email: Annotated[str, Form(description="Contact's email address")],
    message: Annotated[str, Form(description="Contact message")]
):
    """Handle contact form submission."""
    logger.info("Processing contact form", extra={
        "name": name,
        "email": email
    })

    # Process the form data
    # (save to database, send email, etc.)

    return {
        "message": "Contact form submitted successfully",
        "contact_id": "12345"
    }

def lambda_handler(event, context):
    return app.resolve(event, context)

What You Get Automatically
With this simple setup, Powertools automatically provides:

Request Validation: Invalid form data returns proper 422 errors
OpenAPI Schema: Generated schema shows form fields and types
Error Handling: Detailed validation errors for debugging
Content-Type Detection: Automatically parses application/x-www-form-urlencoded

Real-World Examples

User Registration Form:

from typing import Annotated, Optional
from aws_lambda_powertools.event_handler.openapi.params import Form
from pydantic import EmailStr

@app.post("/register")
def register_user(
    username: Annotated[str, Form(min_length=3, max_length=20)],
    email: Annotated[EmailStr, Form()],
    password: Annotated[str, Form(min_length=8)],
    newsletter: Annotated[bool, Form()] = False,
    referral_code: Annotated[Optional[str], Form()] = None
):
    """User registration endpoint."""

    # Automatic validation ensures:
    # - username is 3-20 characters
    # - email is valid format
    # - password is at least 8 characters
    # - newsletter is boolean
    # - referral_code is optional

    return {"user_id": "user_123", "status": "registered"}

Survey/Feedback Form:

from typing import List
from enum import Enum

class SatisfactionLevel(str, Enum):
    VERY_SATISFIED = "very_satisfied"
    SATISFIED = "satisfied" 
    NEUTRAL = "neutral"
    DISSATISFIED = "dissatisfied"
    VERY_DISSATISFIED = "very_dissatisfied"

@app.post("/feedback")
def submit_feedback(
    overall_satisfaction: Annotated[SatisfactionLevel, Form()],
    product_rating: Annotated[int, Form(ge=1, le=5, description="Rating from 1-5")],
    comments: Annotated[str, Form(max_length=1000)],
    recommend: Annotated[bool, Form(description="Would you recommend us?")],
    improvements: Annotated[Optional[str], Form()] = None
):
    """Process customer feedback survey."""

    return {
        "feedback_id": "fb_456",
        "thank_you_message": "Thank you for your feedback!"
    }

Advanced Features

Custom Validation
You can add custom validation using Pydantic validators:

from pydantic import field_validator

class ContactRequest(BaseModel):
    name: Annotated[str, Form()]
    email: Annotated[str, Form()]
    phone: Annotated[str, Form()]

    @field_validator('phone')
    def validate_phone(cls, v):
        # Custom phone validation logic
        if not re.match(r'^\+?[\d\s-()]+$', v):
            raise ValueError('Invalid phone number format')
        return v

@app.post("/contact-advanced")
def advanced_contact(contact: ContactRequest):
    return {"status": "received", "contact_id": "contact_789"}

Error Handling
Form validation errors are automatically formatted:

# When invalid data is sent, you get detailed error responses:
{
    "detail": [
        {
            "type": "string_too_short",
            "loc": ["body", "username"],
            "msg": "String should have at least 3 characters",
            "input": "ab"
        },
        {
            "type": "value_error",
            "loc": ["body", "email"],
            "msg": "Invalid email format"
        }
    ]
}

Testing Your Form Endpoints

Using curl:
# Test the contact form curl -X POST https://your-api.execute-api.region.amazonaws.com/contact \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "name=John Doe&[email protected]&message=Hello World"

Using Python requests:

import requests

response = requests.post(
    'https://your-api.execute-api.region.amazonaws.com/contact',
    data={
        'name': 'Jane Smith',
        'email': '[email protected]', 
        'message': 'Great service!'
    }
)

print(response.json())

HTML Form:

<form action="https://your-api.execute-api.region.amazonaws.com/contact" method="POST">
    <input type="text" name="name" required>
    <input type="email" name="email" required>
    <textarea name="message" required></textarea>
    <button type="submit">Send Message</button>
</form>

OpenAPI Documentation

Your form endpoints automatically generate proper OpenAPI documentation. Here's what the generated schema looks like:

paths:
  /contact:
    post:
      requestBody:
        required: true
        content:
          application/x-www-form-urlencoded:
            schema:
              type: object
              properties:
                name:
                  type: string
                  description: "Contact's full name"
                email:
                  type: string
                  description: "Contact's email address"
                message:
                  type: string
                  description: "Contact message"
              required: ["name", "email", "message"]

Integration with Swagger UI
Enable Swagger UI to get an interactive API explorer:

app = APIGatewayRestResolver(enable_validation=True)
app.enable_swagger(
    path="/docs",
    title="My Contact API",
    version="1.0.0"
)

Deployment Best Practices

Environment Configuration

import os
from aws_lambda_powertools import Logger, Tracer, Metrics

logger = Logger()
tracer = Tracer()
metrics = Metrics()

# Environment-specific settings
DEBUG = os.getenv('DEBUG', 'false').lower() == 'true'
MAX_MESSAGE_LENGTH = int(os.getenv('MAX_MESSAGE_LENGTH', '1000'))

app = APIGatewayRestResolver(
    enable_validation=True,
    debug=DEBUG
)

SAM Template:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  ContactFormFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.11
      Environment:
        Variables:
          POWERTOOLS_SERVICE_NAME: contact-api
          MAX_MESSAGE_LENGTH: 2000
      Events:
        ContactForm:
          Type: Api
          Properties:
            Path: /contact
            Method: post

Monitoring and Observability

Powertools automatically provides structured logging:

@app.post("/contact")
@tracer.capture_method
def submit_contact_form(name: Annotated[str, Form()], email: Annotated[str, Form()]):

    # Add custom metrics
    metrics.add_metric(name="ContactFormSubmission", unit="Count", value=1)
    metrics.add_metadata(key="email_domain", value=email.split('@')[1])

    # Structured logging
    logger.info("Contact form submitted", extra={
        "name": name,
        "email_domain": email.split('@')[1]
    })

    return {"status": "success"}

Security Considerations

Input Sanitization

import html

@app.post("/comment")
def submit_comment(
    content: Annotated[str, Form(max_length=500)]
):
    # Sanitize HTML content
    clean_content = html.escape(content)

    # Additional sanitization as needed
    return {"comment_id": "comment_123"}

Rate Limiting
Use AWS API Gateway throttling or implement custom rate limiting:

from functools import wraps
import time

def rate_limit(max_requests_per_minute=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Implement rate limiting logic
            return func(*args, **kwargs)
        return wrapper
    return decorator

@app.post("/contact")
@rate_limit(max_requests_per_minute=5)
def submit_contact_form(name: Annotated[str, Form()]):
    return {"status": "received"}

Performance Tips