FastAPI Tutorial Part 3: Request Bodies and Pydantic - Data Validation Mastery

Understanding Request Bodies

Request bodies carry data from clients to your API, typically in JSON format. FastAPI uses Pydantic models to automatically parse, validate, and document request data.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    quantity: int = 1

@app.post("/items")
def create_item(item: Item):
    return {"item": item, "total": item.price * item.quantity}

Pydantic Basics

Creating Models

from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
    age: Optional[int] = None
    created_at: datetime = datetime.now()
    tags: List[str] = []

Using Models in Endpoints

@app.post("/users")
def create_user(user: User):
    """
    Request body is automatically:
    - Parsed from JSON
    - Validated against the model
    - Converted to proper Python types
    """
    return {"user": user}

# Example request:
# POST /users
# {
#     "username": "john_doe",
#     "email": "john@example.com",
#     "full_name": "John Doe",
#     "tags": ["developer", "python"]
# }

Field Validation

Basic Field Constraints

from pydantic import BaseModel, Field

class Product(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: str = Field(default="", max_length=1000)
    price: float = Field(gt=0, description="Price must be positive")
    quantity: int = Field(ge=0, le=10000)
    sku: str = Field(pattern=r"^[A-Z]{2}-\d{4}$")

Field Validation Options

Constraint	Type	Description
gt	number	Greater than
ge	number	Greater than or equal
lt	number	Less than
le	number	Less than or equal
multiple_of	number	Must be multiple of
min_length	string	Minimum length
max_length	string	Maximum length
pattern	string	Regex pattern

class Order(BaseModel):
    order_id: str = Field(pattern=r"^ORD-\d{8}$")
    amount: float = Field(gt=0, le=1000000)
    quantity: int = Field(ge=1, le=100, multiple_of=1)
    discount: float = Field(ge=0, le=100, description="Percentage")

Field Metadata

class Article(BaseModel):
    title: str = Field(
        min_length=5,
        max_length=200,
        title="Article Title",
        description="The main title of the article",
        examples=["Getting Started with FastAPI"]
    )
    content: str = Field(
        min_length=100,
        title="Article Content",
        description="The full article text in markdown"
    )
    views: int = Field(
        default=0,
        ge=0,
        title="View Count",
        json_schema_extra={"readOnly": True}
    )

Nested Models

Simple Nesting

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

class Company(BaseModel):
    name: str
    address: Address

class Employee(BaseModel):
    name: str
    email: str
    company: Company

@app.post("/employees")
def create_employee(employee: Employee):
    return employee

# Example request:
# {
#     "name": "Alice Smith",
#     "email": "alice@company.com",
#     "company": {
#         "name": "Tech Corp",
#         "address": {
#             "street": "123 Main St",
#             "city": "San Francisco",
#             "country": "USA",
#             "postal_code": "94102"
#         }
#     }
# }

Lists of Models

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(ge=1)
    unit_price: float = Field(gt=0)

class Order(BaseModel):
    customer_id: int
    items: List[OrderItem] = Field(min_length=1)
    notes: Optional[str] = None

    @property
    def total(self) -> float:
        return sum(item.quantity * item.unit_price for item in self.items)

@app.post("/orders")
def create_order(order: Order):
    return {
        "order": order,
        "total": order.total,
        "item_count": len(order.items)
    }

Self-Referencing Models

from typing import Optional, List, ForwardRef

class Category(BaseModel):
    name: str
    parent: Optional["Category"] = None
    children: List["Category"] = []

# Required for self-referencing
Category.model_rebuild()

@app.post("/categories")
def create_category(category: Category):
    return category

Custom Validators

Field Validators

from pydantic import BaseModel, field_validator

class User(BaseModel):
    username: str
    email: str
    password: str

    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("Username must be alphanumeric")
        return v.lower()

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v.lower()

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain a digit")
        return v

Model Validators

from pydantic import BaseModel, model_validator

class DateRange(BaseModel):
    start_date: datetime
    end_date: datetime

    @model_validator(mode="after")
    def check_dates(self):
        if self.end_date <= self.start_date:
            raise ValueError("end_date must be after start_date")
        return self

class PriceRange(BaseModel):
    min_price: float
    max_price: float

    @model_validator(mode="after")
    def check_prices(self):
        if self.max_price < self.min_price:
            raise ValueError("max_price must be >= min_price")
        return self

Before Validation

class Product(BaseModel):
    name: str
    tags: List[str]

    @field_validator("name", mode="before")
    @classmethod
    def strip_name(cls, v):
        if isinstance(v, str):
            return v.strip()
        return v

    @field_validator("tags", mode="before")
    @classmethod
    def split_tags(cls, v):
        if isinstance(v, str):
            return [t.strip() for t in v.split(",")]
        return v

# Accepts both:
# {"name": "  Product  ", "tags": ["a", "b"]}
# {"name": "  Product  ", "tags": "a, b, c"}

Special Types

Email and URLs

from pydantic import BaseModel, EmailStr, HttpUrl, AnyUrl

class Contact(BaseModel):
    email: EmailStr
    website: HttpUrl
    callback_url: AnyUrl

# Install email-validator: pip install email-validator

UUIDs

from uuid import UUID, uuid4

class Resource(BaseModel):
    id: UUID = Field(default_factory=uuid4)
    name: str

@app.post("/resources")
def create_resource(resource: Resource):
    return resource

Dates and Times

from datetime import datetime, date, time, timedelta

class Event(BaseModel):
    name: str
    event_date: date
    start_time: time
    end_time: time
    created_at: datetime = Field(default_factory=datetime.now)
    duration: timedelta = timedelta(hours=1)

Constrained Types

from pydantic import BaseModel, constr, conint, confloat, conlist

class StrictModel(BaseModel):
    # Constrained string
    code: constr(min_length=5, max_length=10, pattern=r"^[A-Z]+$")

    # Constrained integer
    quantity: conint(ge=1, le=100)

    # Constrained float
    rating: confloat(ge=0, le=5)

    # Constrained list
    tags: conlist(str, min_length=1, max_length=5)

Model Configuration

Configure Model Behavior

from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,  # Strip whitespace from strings
        str_min_length=1,  # Minimum string length
        validate_assignment=True,  # Validate on attribute assignment
        extra="forbid",  # Forbid extra fields
        frozen=False,  # Allow mutation
    )

    username: str
    email: str

Extra Fields Handling

# Ignore extra fields (default)
class UserIgnoreExtra(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str

# Forbid extra fields
class UserForbidExtra(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str

# Allow extra fields
class UserAllowExtra(BaseModel):
    model_config = ConfigDict(extra="allow")
    name: str

Aliases

class User(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    user_name: str = Field(alias="userName")
    email_address: str = Field(alias="emailAddress")

# Accepts both:
# {"userName": "john", "emailAddress": "john@example.com"}
# {"user_name": "john", "email_address": "john@example.com"}

Multiple Body Parameters

Combining Body Parameters

from fastapi import Body

class Item(BaseModel):
    name: str
    price: float

class User(BaseModel):
    username: str

@app.post("/items")
def create_item_with_user(item: Item, user: User):
    """
    Expects:
    {
        "item": {"name": "Widget", "price": 9.99},
        "user": {"username": "john"}
    }
    """
    return {"item": item, "user": user}

Singular Body Values

@app.post("/items")
def create_item(
    item: Item,
    importance: int = Body(gt=0)
):
    """
    Expects:
    {
        "item": {"name": "Widget", "price": 9.99},
        "importance": 5
    }
    """
    return {"item": item, "importance": importance}

Embed Single Model

@app.post("/items")
def create_item(item: Item = Body(embed=True)):
    """
    Expects:
    {
        "item": {"name": "Widget", "price": 9.99}
    }
    Instead of just:
    {"name": "Widget", "price": 9.99}
    """
    return item

Practical Example: Blog API

from fastapi import FastAPI, Body, HTTPException
from pydantic import BaseModel, Field, field_validator, EmailStr
from typing import Optional, List
from datetime import datetime
from enum import Enum

app = FastAPI()

class PostStatus(str, Enum):
    draft = "draft"
    published = "published"
    archived = "archived"

class Author(BaseModel):
    name: str = Field(min_length=2, max_length=100)
    email: EmailStr
    bio: Optional[str] = Field(default=None, max_length=500)

class Tag(BaseModel):
    name: str = Field(min_length=2, max_length=30)
    slug: str = Field(pattern=r"^[a-z0-9-]+$")

class Comment(BaseModel):
    author_name: str = Field(min_length=2, max_length=50)
    author_email: EmailStr
    content: str = Field(min_length=10, max_length=2000)
    created_at: datetime = Field(default_factory=datetime.now)

class PostCreate(BaseModel):
    title: str = Field(min_length=5, max_length=200)
    slug: str = Field(pattern=r"^[a-z0-9-]+$")
    content: str = Field(min_length=100)
    excerpt: Optional[str] = Field(default=None, max_length=300)
    author: Author
    tags: List[Tag] = Field(default=[], max_length=10)
    status: PostStatus = PostStatus.draft
    featured_image: Optional[str] = None

    @field_validator("slug")
    @classmethod
    def slug_lowercase(cls, v: str) -> str:
        return v.lower()

    @field_validator("excerpt", mode="before")
    @classmethod
    def generate_excerpt(cls, v, info):
        if v is None and "content" in info.data:
            content = info.data["content"]
            return content[:297] + "..." if len(content) > 300 else content
        return v

class PostUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=5, max_length=200)
    content: Optional[str] = Field(default=None, min_length=100)
    excerpt: Optional[str] = Field(default=None, max_length=300)
    tags: Optional[List[Tag]] = None
    status: Optional[PostStatus] = None

class Post(PostCreate):
    id: int
    created_at: datetime
    updated_at: datetime
    comments: List[Comment] = []
    view_count: int = 0

# In-memory storage
posts_db: dict[int, Post] = {}
post_counter = 0

@app.post("/posts", response_model=Post, status_code=201)
def create_post(post: PostCreate):
    """Create a new blog post."""
    global post_counter
    post_counter += 1

    new_post = Post(
        id=post_counter,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        **post.model_dump()
    )
    posts_db[post_counter] = new_post
    return new_post

@app.put("/posts/{post_id}", response_model=Post)
def update_post(post_id: int, post_update: PostUpdate):
    """Update an existing blog post."""
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")

    stored_post = posts_db[post_id]
    update_data = post_update.model_dump(exclude_unset=True)

    for field, value in update_data.items():
        setattr(stored_post, field, value)

    stored_post.updated_at = datetime.now()
    return stored_post

@app.post("/posts/{post_id}/comments", response_model=Comment)
def add_comment(post_id: int, comment: Comment):
    """Add a comment to a post."""
    if post_id not in posts_db:
        raise HTTPException(status_code=404, detail="Post not found")

    posts_db[post_id].comments.append(comment)
    return comment

@app.post("/posts/bulk", response_model=List[Post])
def create_bulk_posts(posts: List[PostCreate] = Body(min_length=1, max_length=10)):
    """Create multiple posts at once."""
    created = []
    global post_counter

    for post in posts:
        post_counter += 1
        new_post = Post(
            id=post_counter,
            created_at=datetime.now(),
            updated_at=datetime.now(),
            **post.model_dump()
        )
        posts_db[post_counter] = new_post
        created.append(new_post)

    return created

Error Responses

FastAPI returns detailed validation errors:

# Invalid request:
# POST /posts
# {"title": "Hi", "content": "short", "author": {"name": "A", "email": "invalid"}}

# Response (422 Unprocessable Entity):
{
    "detail": [
        {
            "type": "string_too_short",
            "loc": ["body", "title"],
            "msg": "String should have at least 5 characters",
            "input": "Hi",
            "ctx": {"min_length": 5}
        },
        {
            "type": "string_too_short",
            "loc": ["body", "content"],
            "msg": "String should have at least 100 characters",
            "input": "short",
            "ctx": {"min_length": 100}
        },
        {
            "type": "value_error",
            "loc": ["body", "author", "email"],
            "msg": "value is not a valid email address",
            "input": "invalid"
        }
    ]
}

Summary

Concept	Description
BaseModel	Base class for Pydantic models
Field()	Field configuration and validation
@field_validator	Custom field validation
@model_validator	Cross-field validation
Body()	Body parameter configuration
ConfigDict	Model behavior configuration

Validation	Usage
Numeric bounds	Field(gt=0, le=100)
String length	Field(min_length=1, max_length=100)
Regex pattern	Field(pattern=r"^[a-z]+$")
Custom logic	@field_validator decorator
Cross-field	@model_validator decorator

Next Steps

In Part 4, we’ll explore Response Models and Status Codes - learning how to control API responses, filter output data, and use proper HTTP status codes.

Series Navigation:

• Part 1: Introduction and Setup
• Part 2: Path and Query Parameters
• Part 3: Request Bodies and Pydantic (You are here)
• Part 4: Response Models and Status Codes