API Models with Pydantic

Pydantic is the foundation for building type-safe API models. This guide demonstrates patterns for REST APIs, FastAPI integration, and real-world API development.

Basic Request and Response Models

Define clear contracts for API endpoints:

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

class UserCreate(BaseModel):
    """Request model for creating a user"""
    username: str = Field(min_length=3, max_length=50)
    email: str
    full_name: str
    password: str = Field(min_length=8)

class UserResponse(BaseModel):
    """Response model for user data"""
    id: int
    username: str
    email: str
    full_name: str
    created_at: datetime
    is_active: bool = True
    
    # Exclude password from response
    model_config = {'from_attributes': True}

class UserUpdate(BaseModel):
    """Request model for updating user - all fields optional"""
    username: Optional[str] = Field(None, min_length=3, max_length=50)
    email: Optional[str] = None
    full_name: Optional[str] = None
    is_active: Optional[bool] = None

FastAPI Integration

Use Pydantic models seamlessly with FastAPI:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, EmailStr
from typing import List

app = FastAPI()

# Models
class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = None
    price: float = Field(..., gt=0)
    tax: float | None = Field(None, ge=0)

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    total: float

# Endpoints
@app.post("/items/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    """Create a new item - request validated automatically"""
    total = item.price + (item.tax or 0)
    return ItemResponse(
        id=1,
        name=item.name,
        price=item.price,
        total=total
    )

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
    """Get item by ID - response validated automatically"""
    # Database lookup simulation
    if item_id == 1:
        return ItemResponse(id=1, name="Widget", price=99.99, total=109.99)
    raise HTTPException(status_code=404, detail="Item not found")

@app.get("/items/", response_model=List[ItemResponse])
async def list_items():
    """List all items"""
    return [
        ItemResponse(id=1, name="Widget", price=99.99, total=109.99),
        ItemResponse(id=2, name="Gadget", price=49.99, total=54.99),
    ]

FastAPI automatically:

Validates request bodies against your Pydantic models
Returns validation errors as HTTP 422 responses
Generates OpenAPI documentation from your models
Serializes responses using your response models

Nested API Models

Handle complex nested data structures:

from pydantic import BaseModel, Field, HttpUrl
from typing import List
from datetime import datetime

class Tag(BaseModel):
    name: str
    color: str = Field(pattern=r'^#[0-9A-Fa-f]{6}$')

class Author(BaseModel):
    id: int
    name: str
    email: str

class Comment(BaseModel):
    id: int
    author: Author
    content: str
    created_at: datetime

class ArticleBase(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    content: str
    tags: List[Tag] = []

class ArticleCreate(ArticleBase):
    """Request model for creating articles"""
    pass

class ArticleResponse(ArticleBase):
    """Response model with computed fields"""
    id: int
    author: Author
    comments: List[Comment] = []
    created_at: datetime
    updated_at: datetime
    view_count: int = 0
    
    model_config = {'from_attributes': True}

Pagination Models

from pydantic import BaseModel, Field
from typing import List, Generic, TypeVar

T = TypeVar('T')

class PaginationParams(BaseModel):
    """Query parameters for pagination"""
    offset: int = Field(0, ge=0)
    limit: int = Field(10, ge=1, le=100)

class PaginatedResponse(BaseModel, Generic[T]):
    """Generic paginated response"""
    items: List[T]
    total: int
    offset: int
    limit: int
    
    @property
    def has_next(self) -> bool:
        return self.offset + self.limit < self.total

# Usage
class Product(BaseModel):
    id: int
    name: str
    price: float

def get_products(pagination: PaginationParams) -> PaginatedResponse[Product]:
    # Database query...
    products = [Product(id=1, name="Widget", price=99.99)]
    return PaginatedResponse(
        items=products,
        total=100,
        offset=pagination.offset,
        limit=pagination.limit
    )

Error Response Models

Standardize error responses:

from pydantic import BaseModel, Field
from typing import List, Optional, Any
from enum import Enum

class ErrorType(str, Enum):
    VALIDATION_ERROR = "validation_error"
    NOT_FOUND = "not_found"
    AUTHENTICATION_ERROR = "authentication_error"
    PERMISSION_DENIED = "permission_denied"
    INTERNAL_ERROR = "internal_error"

class ErrorDetail(BaseModel):
    """Individual error detail"""
    field: Optional[str] = None
    message: str
    code: str

class ErrorResponse(BaseModel):
    """Standard error response"""
    error: ErrorType
    message: str
    details: List[ErrorDetail] = []
    request_id: Optional[str] = None

# Usage in FastAPI
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError

app = FastAPI()

@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    details = [
        ErrorDetail(
            field='.'.join(str(loc) for loc in error['loc']),
            message=error['msg'],
            code=error['type']
        )
        for error in exc.errors()
    ]
    
    error_response = ErrorResponse(
        error=ErrorType.VALIDATION_ERROR,
        message="Validation failed",
        details=details
    )
    
    return JSONResponse(
        status_code=422,
        content=error_response.model_dump()
    )

Field Aliases for APIs

Map between internal and external field names:

from pydantic import BaseModel, Field

class UserAPI(BaseModel):
    """API model with camelCase for JavaScript compatibility"""
    user_id: int = Field(alias='userId')
    first_name: str = Field(alias='firstName')
    last_name: str = Field(alias='lastName')
    email_address: str = Field(alias='emailAddress')
    
    model_config = {
        'populate_by_name': True  # Accept both snake_case and camelCase
    }

# Accepts camelCase from API
user = UserAPI.model_validate({
    'userId': 1,
    'firstName': 'John',
    'lastName': 'Doe',
    'emailAddress': '[email protected]'
})

# Outputs with aliases
print(user.model_dump(by_alias=True))
# {'userId': 1, 'firstName': 'John', 'lastName': 'Doe', 'emailAddress': '[email protected]'}

Use populate_by_name=True to accept both the field name and its alias. This is useful during API migrations when clients may use either naming convention.

JSON Validation and Parsing

Validate and parse JSON data directly:

import httpx
from pydantic import BaseModel, EmailStr
from typing import List

class User(BaseModel):
    id: int
    name: str
    email: EmailStr

class UsersResponse(BaseModel):
    users: List[User]
    total: int

# Parse JSON string
json_data = '''
{
    "users": [
        {"id": 1, "name": "Alice", "email": "[email protected]"},
        {"id": 2, "name": "Bob", "email": "[email protected]"}
    ],
    "total": 2
}
'''

response = UsersResponse.model_validate_json(json_data)
print(response.users[0].name)  # Alice

# Parse HTTP response
async def fetch_users():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/users')
        response.raise_for_status()
        return UsersResponse.model_validate(response.json())

Query Parameter Models

Validate complex query parameters:

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import date
from enum import Enum

class SortOrder(str, Enum):
    ASC = "asc"
    DESC = "desc"

class ProductFilter(BaseModel):
    """Query parameters for filtering products"""
    category: Optional[str] = None
    min_price: Optional[float] = Field(None, ge=0)
    max_price: Optional[float] = Field(None, ge=0)
    in_stock: Optional[bool] = None
    tags: List[str] = Field(default_factory=list)
    created_after: Optional[date] = None
    sort_by: str = "created_at"
    sort_order: SortOrder = SortOrder.DESC
    
    @property
    def has_filters(self) -> bool:
        return any([
            self.category,
            self.min_price is not None,
            self.max_price is not None,
            self.in_stock is not None,
            self.tags,
            self.created_after
        ])

# FastAPI usage
from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/products/")
async def search_products(filters: ProductFilter = Depends()):
    # filters is automatically validated from query parameters
    if filters.has_filters:
        # Apply filters to database query
        pass
    return {"filters": filters.model_dump(exclude_none=True)}

Webhook Models

Step 1: Define Event Types

from pydantic import BaseModel, Field
from typing import Literal, Union
from datetime import datetime

class WebhookBase(BaseModel):
    event_id: str
    timestamp: datetime

class OrderCreatedEvent(WebhookBase):
    event_type: Literal["order.created"]
    order_id: int
    customer_id: int
    total: float

class OrderShippedEvent(WebhookBase):
    event_type: Literal["order.shipped"]
    order_id: int
    tracking_number: str
    carrier: str

class PaymentReceivedEvent(WebhookBase):
    event_type: Literal["payment.received"]
    payment_id: int
    amount: float
    currency: str

Step 2: Create Union Type

from typing import Annotated
from pydantic import Field, Discriminator

WebhookEvent = Annotated[
    Union[OrderCreatedEvent, OrderShippedEvent, PaymentReceivedEvent],
    Field(discriminator='event_type')
]

class WebhookPayload(BaseModel):
    event: WebhookEvent
    signature: str

Step 3: Handle Webhooks

from fastapi import FastAPI, Header, HTTPException
import hmac
import hashlib

app = FastAPI()

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)

@app.post("/webhooks/")
async def handle_webhook(
    payload: WebhookPayload,
    x_signature: str = Header(...)
):
    # Verify signature
    # if not verify_signature(...):
    #     raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Handle different event types
    if payload.event.event_type == "order.created":
        print(f"New order: {payload.event.order_id}")
    elif payload.event.event_type == "order.shipped":
        print(f"Order shipped: {payload.event.tracking_number}")
    elif payload.event.event_type == "payment.received":
        print(f"Payment received: {payload.event.amount}")
    
    return {"status": "processed"}

File Upload Models

from pydantic import BaseModel, Field, field_validator
from fastapi import FastAPI, UploadFile, File, Form
from typing import List
import mimetypes

class FileMetadata(BaseModel):
    filename: str
    content_type: str
    size: int = Field(gt=0)
    
    @field_validator('content_type')
    @classmethod
    def validate_content_type(cls, v: str) -> str:
        allowed = ['image/jpeg', 'image/png', 'image/gif', 'application/pdf']
        if v not in allowed:
            raise ValueError(f'Content type {v} not allowed')
        return v

class UploadResponse(BaseModel):
    files: List[FileMetadata]
    total_size: int

app = FastAPI()

@app.post("/upload/", response_model=UploadResponse)
async def upload_files(
    files: List[UploadFile] = File(...),
    description: str = Form(None)
):
    metadata_list = []
    total_size = 0
    
    for file in files:
        content = await file.read()
        size = len(content)
        
        metadata = FileMetadata(
            filename=file.filename,
            content_type=file.content_type,
            size=size
        )
        metadata_list.append(metadata)
        total_size += size
    
    return UploadResponse(files=metadata_list, total_size=total_size)

Rate Limiting Headers

from pydantic import BaseModel, Field
from datetime import datetime, timedelta
from typing import Optional

class RateLimitInfo(BaseModel):
    limit: int = Field(description="Maximum requests allowed")
    remaining: int = Field(description="Requests remaining")
    reset: datetime = Field(description="When limit resets")
    
    def to_headers(self) -> dict:
        """Convert to response headers"""
        return {
            'X-RateLimit-Limit': str(self.limit),
            'X-RateLimit-Remaining': str(self.remaining),
            'X-RateLimit-Reset': str(int(self.reset.timestamp()))
        }

# FastAPI middleware
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Check rate limit...
        rate_limit = RateLimitInfo(
            limit=100,
            remaining=95,
            reset=datetime.now() + timedelta(hours=1)
        )
        
        response = await call_next(request)
        
        # Add headers
        for key, value in rate_limit.to_headers().items():
            response.headers[key] = value
        
        return response

Summary

API models with Pydantic provide:

Type-safe request and response validation
Automatic FastAPI integration with docs generation
Nested models for complex data structures
Pagination patterns (offset and cursor-based)
Standardized error responses
Field aliases for API compatibility
JSON validation and parsing
Query parameter validation
Webhook event handling with discriminated unions
File upload validation
Response header models

These patterns create robust, self-documenting APIs with comprehensive validation.

Get Started

Core Concepts

Advanced Features

Guides

Examples

Integrations

API Models

API Models with Pydantic

Basic Request and Response Models

FastAPI Integration

Nested API Models

Error Response Models

Field Aliases for APIs

JSON Validation and Parsing

Query Parameter Models

Webhook Models

File Upload Models

Rate Limiting Headers

Summary

Build docs developers (and LLMs) love

Get Started

Core Concepts

Advanced Features

Guides

Examples

Integrations

​API Models with Pydantic

​Basic Request and Response Models

​FastAPI Integration

​Nested API Models

​Pagination Models

​Error Response Models

​Field Aliases for APIs

​JSON Validation and Parsing

​Query Parameter Models

​Webhook Models

​File Upload Models

​Rate Limiting Headers

​Summary

Build docs developers (and LLMs) love

API Models with Pydantic

Basic Request and Response Models

FastAPI Integration

Nested API Models

Pagination Models

Error Response Models

Field Aliases for APIs

JSON Validation and Parsing

Query Parameter Models

Webhook Models

File Upload Models

Rate Limiting Headers

Summary