FastAPI Request Body Validation - Complete Guide

FastAPI's request body validation with Pydantic provides powerful data validation capabilities. This guide covers advanced validation patterns and error handling.

1. Basic Request Body Validation

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from fastapi import FastAPI, HTTPException

app = FastAPI()

class UserCreate(BaseModel):
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    name: str = Field(..., min_length=2, max_length=50)
    age: int = Field(..., ge=18, le=120)
    password: str = Field(..., min_length=8)
    confirm_password: str

    @validator('confirm_password')
    def passwords_match(cls, v, values):
        if 'password' in values and v != values['password']:
            raise ValueError('passwords do not match')
        return v

@app.post("/users/")
async def create_user(user: UserCreate):
    # Password confirmation field is not included in the response
    user_dict = user.dict(exclude={'confirm_password'})
    return {"message": "User created", "user": user_dict}

2. Nested Model Validation

class Address(BaseModel):
    street: str = Field(..., min_length=5, max_length=100)
    city: str = Field(..., min_length=2, max_length=50)
    state: str = Field(..., min_length=2, max_length=50)
    zip_code: str = Field(..., regex=r'^\d{5}(-\d{4})?$')
    country: str = Field(default="US", max_length=2)

class UserProfile(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    phone: Optional[str] = Field(None, regex=r'^\+?1?\d{9,15}$')
    address: Address
    emergency_contacts: List[str] = Field(default_factory=list, max_items=3)

    @validator('emergency_contacts')
    def validate_emergency_contacts(cls, v):
        for contact in v:
            if not contact.strip():
                raise ValueError('Emergency contact cannot be empty')
        return v

@app.post("/profiles/")
async def create_profile(profile: UserProfile):
    return {"message": "Profile created", "profile": profile}

3. Custom Validation Methods

from pydantic import BaseModel, validator, root_validator
import re

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=3, max_length=100)
    price: float = Field(..., gt=0)
    category: str
    tags: List[str] = Field(default_factory=list)
    sku: str = Field(..., min_length=6, max_length=20)

    @validator('name')
    def validate_name(cls, v):
        if not re.match(r'^[a-zA-Z0-9\s\-_]+$', v):
            raise ValueError('Name contains invalid characters')
        return v.title()

    @validator('sku')
    def validate_sku(cls, v):
        if not re.match(r'^[A-Z]{2}\d{4,}$', v):
            raise ValueError('SKU must start with 2 letters followed by at least 4 digits')
        return v.upper()

    @validator('tags', each_item=True)
    def validate_tags(cls, v):
        if len(v) < 2:
            raise ValueError('Each tag must be at least 2 characters long')
        return v.lower()

    @root_validator
    def validate_category_price(cls, values):
        category = values.get('category')
        price = values.get('price')

        if category == 'premium' and price < 100:
            raise ValueError('Premium products must be priced at least $100')

        return values

@app.post("/products/")
async def create_product(product: ProductCreate):
    return {"message": "Product created", "product": product}

4. Conditional Validation

from typing import Union, Literal

class PaymentMethod(BaseModel):
    type: Literal['credit_card', 'bank_transfer', 'paypal']

    # Credit card fields
    card_number: Optional[str] = None
    cvv: Optional[str] = None
    expiry_month: Optional[int] = None
    expiry_year: Optional[int] = None

    # Bank transfer fields
    account_number: Optional[str] = None
    routing_number: Optional[str] = None

    # PayPal fields
    paypal_email: Optional[str] = None

    @root_validator
    def validate_payment_fields(cls, values):
        payment_type = values.get('type')

        if payment_type == 'credit_card':
            required_fields = ['card_number', 'cvv', 'expiry_month', 'expiry_year']
            for field in required_fields:
                if not values.get(field):
                    raise ValueError(f'{field} is required for credit card payments')

            # Validate credit card number (simplified)
            card_number = values.get('card_number', '').replace(' ', '')
            if not re.match(r'^\d{13,19}$', card_number):
                raise ValueError('Invalid credit card number')

        elif payment_type == 'bank_transfer':
            required_fields = ['account_number', 'routing_number']
            for field in required_fields:
                if not values.get(field):
                    raise ValueError(f'{field} is required for bank transfers')

        elif payment_type == 'paypal':
            if not values.get('paypal_email'):
                raise ValueError('PayPal email is required for PayPal payments')

        return values

@app.post("/payments/")
async def process_payment(payment: PaymentMethod):
    return {"message": "Payment processed", "payment_type": payment.type}

5. File Upload Validation

from fastapi import File, UploadFile, Form, HTTPException
from pydantic import BaseModel, validator
import magic

class FileUploadData(BaseModel):
    title: str = Field(..., min_length=3, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    category: str

    @validator('title')
    def validate_title(cls, v):
        if not v.strip():
            raise ValueError('Title cannot be empty')
        return v.strip()

@app.post("/upload/")
async def upload_file(
    file: UploadFile = File(...),
    data: str = Form(...)
):
    # Parse the form data
    try:
        file_data = FileUploadData.parse_raw(data)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # Validate file
    if file.size > 10 * 1024 * 1024:  # 10MB limit
        raise HTTPException(status_code=413, detail="File too large")

    # Validate file type
    content = await file.read()
    file_type = magic.from_buffer(content, mime=True)

    allowed_types = ['image/jpeg', 'image/png', 'application/pdf']
    if file_type not in allowed_types:
        raise HTTPException(
            status_code=415, 
            detail=f"File type {file_type} not allowed"
        )

    # Reset file pointer
    await file.seek(0)

    return {
        "message": "File uploaded successfully",
        "filename": file.filename,
        "size": file.size,
        "type": file_type,
        "data": file_data
    }

6. Advanced Validation with Dependencies

from fastapi import Depends
from sqlalchemy.orm import Session
from app.database import get_db

class UserUpdate(BaseModel):
    email: Optional[str] = None
    name: Optional[str] = None

    @validator('email')
    def validate_email(cls, v):
        if v and not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', v):
            raise ValueError('Invalid email format')
        return v

async def validate_unique_email(user_id: int, user_update: UserUpdate, db: Session = Depends(get_db)):
    if user_update.email:
        existing_user = db.query(User).filter(
            User.email == user_update.email,
            User.id != user_id
        ).first()

        if existing_user:
            raise HTTPException(
                status_code=400,
                detail="Email already registered"
            )

    return user_update

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    user_update: UserUpdate = Depends(validate_unique_email),
    db: Session = Depends(get_db)
):
    # Update user logic here
    return {"message": "User updated", "user_id": user_id}

7. Custom Validation Error Handling

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import ValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field_path = " -> ".join(str(loc) for loc in error["loc"][1:])  # Skip 'body'

        errors.append({
            "field": field_path,
            "message": error["msg"],
            "invalid_value": error.get("input"),
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": "Validation failed",
            "details": errors,
            "total_errors": len(errors)
        }
    )

# Custom validator for complex business rules
class OrderCreate(BaseModel):
    items: List[dict] = Field(..., min_items=1, max_items=50)
    customer_id: int
    shipping_address: Address
    payment_method: PaymentMethod
    discount_code: Optional[str] = None

    @validator('items')
    def validate_items(cls, v):
        total_value = 0
        for item in v:
            if 'quantity' not in item or 'price' not in item:
                raise ValueError('Each item must have quantity and price')

            quantity = item.get('quantity', 0)
            price = item.get('price', 0)

            if quantity <= 0:
                raise ValueError('Item quantity must be positive')
            if price <= 0:
                raise ValueError('Item price must be positive')

            total_value += quantity * price

        if total_value > 10000:
            raise ValueError('Order total cannot exceed $10,000')

        return v

@app.post("/orders/")
async def create_order(order: OrderCreate):
    return {"message": "Order created", "order": order}

8. Request Body with Query Parameters

class SearchFilters(BaseModel):
    category: Optional[str] = None
    min_price: Optional[float] = Field(None, ge=0)
    max_price: Optional[float] = Field(None, ge=0)
    tags: List[str] = Field(default_factory=list)

    @root_validator
    def validate_price_range(cls, values):
        min_price = values.get('min_price')
        max_price = values.get('max_price')

        if min_price is not None and max_price is not None:
            if min_price > max_price:
                raise ValueError('min_price cannot be greater than max_price')

        return values

@app.post("/search/")
async def search_products(
    filters: SearchFilters,
    page: int = Field(1, ge=1),
    limit: int = Field(10, ge=1, le=100)
):
    # Search logic here
    return {
        "filters": filters,
        "pagination": {"page": page, "limit": limit}
    }

Best Practices

1. Validation Performance

• Use validator for field-level validation
• Use root_validator for cross-field validation
• Cache compiled regex patterns
• Validate expensive operations asynchronously

2. Error Messages

• Provide clear, actionable error messages
• Include field names and expected formats
• Avoid exposing internal implementation details

3. Security Considerations

• Validate file uploads thoroughly
• Sanitize string inputs
• Limit request body sizes
• Validate against SQL injection patterns

Related Guides

• FastAPI Pydantic Validation
• FastAPI Custom Exception Handling
• FastAPI CORS with React