Project 3: API Request/Response Validator

Project 3: API Request/Response Validator

Build a complete REST API with Pydantic models for request validation, response serialization, and automatic OpenAPI documentation—demonstrating best practices for production APIs.

---

Learning Objectives

By completing this project, you will:

1. Understand FastAPI + Pydantic synergy - How they work together seamlessly
2. Master request validation - Validate body, query, path, and header parameters
3. Design response models - Separate input schemas from output schemas
4. Generate OpenAPI documentation - Leverage Field metadata for rich docs
5. Handle validation errors gracefully - Custom error responses and handlers
6. Implement CRUD patterns - Standard API patterns with proper typing

---

Theoretical Foundation

Why FastAPI + Pydantic?

FastAPI was built around Pydantic. They share a creator (Sebastián Ramírez) and are designed to work together:

┌─────────────────────────────────────────────────────────────────┐
│                       HTTP Request                               │
│                                                                  │
│  POST /users                                                     │
│  Content-Type: application/json                                  │
│  X-Request-ID: abc123                                           │
│                                                                  │
│  {                                                               │
│    "name": "John Doe",                                          │
│    "email": "john@example.com",                                 │
│    "age": 30                                                    │
│  }                                                               │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    FastAPI + Pydantic                            │
│                                                                  │
│  @app.post("/users", response_model=UserResponse)               │
│  async def create_user(                                         │
│      user: UserCreate,  ◄─── Body validated by Pydantic        │
│      x_request_id: str = Header(...),  ◄─── Header validated   │
│  ):                                                              │
│      # user is already validated!                               │
│      return save_user(user)  ◄─── Response validated            │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      HTTP Response                               │
│                                                                  │
│  201 Created                                                     │
│  Content-Type: application/json                                  │
│                                                                  │
│  {                                                               │
│    "id": 1,                                                     │
│    "name": "John Doe",                                          │
│    "email": "john@example.com",                                 │
│    "created_at": "2024-01-15T10:30:00Z"                         │
│  }   # Note: age excluded, id and created_at added              │
└─────────────────────────────────────────────────────────────────┘

Key Benefits:

1. Automatic Validation - Request data is validated before your code runs
2. Type Safety - Your function receives typed objects, not dicts
3. Documentation - OpenAPI/Swagger docs generated automatically
4. Serialization - Responses are serialized according to your models
5. Editor Support - Full autocomplete and type checking

The Request Validation Pipeline

When a request hits a FastAPI endpoint, it goes through several validation stages:

1. Path Parameters    →  /users/{user_id}     →  int validation
2. Query Parameters   →  ?page=1&limit=10     →  Pydantic types
3. Headers           →  X-API-Key: abc        →  String/custom types
4. Cookies           →  session=xyz           →  String/custom types
5. Request Body      →  JSON payload          →  Pydantic model
6. Form Data         →  multipart/form-data   →  Pydantic model

Each parameter can have:

• Type annotation - user_id: int
• Default value - page: int = 1
• Validation - page: int = Query(ge=1)
• Description - page: int = Query(..., description="Page number")

Separating Input and Output Models

A critical pattern in API design: never use the same model for input and output.

# BAD: Same model for everything
class User(BaseModel):
    id: int                    # Generated by database
    name: str
    email: str
    password: str              # NEVER return this!
    created_at: datetime       # Generated by database
    hashed_password: str       # Internal

# GOOD: Separate models for different purposes
class UserBase(BaseModel):
    """Shared fields"""
    name: str
    email: EmailStr

class UserCreate(UserBase):
    """For creating users - includes password"""
    password: str = Field(min_length=8)

class UserUpdate(BaseModel):
    """For updating users - all fields optional"""
    name: Optional[str] = None
    email: Optional[EmailStr] = None

class UserResponse(UserBase):
    """For returning users - includes id, excludes password"""
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

Why separate models?

Model	Password	ID	created_at	All Required
UserCreate	✓ Include	✗ No	✗ No	✓ Yes
UserUpdate	✗ Optional	✗ No	✗ No	✗ No
UserResponse	✗ Never	✓ Yes	✓ Yes	✓ Yes

OpenAPI Schema Generation

FastAPI automatically generates OpenAPI (formerly Swagger) documentation from your Pydantic models:

class User(BaseModel):
    name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="User's full name",
        examples=["John Doe", "Jane Smith"]
    )
    email: EmailStr = Field(
        ...,
        description="User's email address",
        examples=["user@example.com"]
    )
    age: int = Field(
        ...,
        ge=0,
        le=150,
        description="User's age in years"
    )

This generates OpenAPI schema:

{
  "User": {
    "type": "object",
    "required": ["name", "email", "age"],
    "properties": {
      "name": {
        "type": "string",
        "minLength": 1,
        "maxLength": 100,
        "description": "User's full name",
        "examples": ["John Doe", "Jane Smith"]
      },
      "email": {
        "type": "string",
        "format": "email",
        "description": "User's email address",
        "examples": ["user@example.com"]
      },
      "age": {
        "type": "integer",
        "minimum": 0,
        "maximum": 150,
        "description": "User's age in years"
      }
    }
  }
}

Error Handling in FastAPI

When validation fails, FastAPI raises RequestValidationError:

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
            "input": error.get("input")
        })

    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation failed",
            "errors": errors
        }
    )

Response Model Filtering

FastAPI uses response_model to:

1. Validate response data
2. Filter out extra fields
3. Generate OpenAPI response schema

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    # Even if save_user returns a dict with 'password',
    # the response will only include UserResponse fields
    db_user = save_user(user)
    return db_user  # password automatically excluded

Advanced response model options:

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    ...

@app.get(
    "/users",
    response_model=List[UserResponse],
    response_model_exclude={"email"},  # Exclude from all
    response_model_include={"id", "name"},  # Only these
)
async def list_users():
    ...

ORM Compatibility with from_attributes

To return SQLAlchemy/ORM objects directly, enable from_attributes:

from pydantic import ConfigDict

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

    model_config = ConfigDict(from_attributes=True)

# Now this works:
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = db.query(User).get(user_id)  # SQLAlchemy model
    return user  # Automatically converted to UserResponse

---

Project Specification

Functional Requirements

Build a REST API for a task management system that demonstrates:

1. Full CRUD operations with proper Pydantic models
2. Nested resources (users have tasks)
3. Pagination and filtering with query parameters
4. Rich error responses with field-level details
5. Comprehensive OpenAPI documentation

API Endpoints

Users:
  POST   /users                 Create a new user
  GET    /users                 List users (paginated)
  GET    /users/{user_id}       Get a single user
  PATCH  /users/{user_id}       Update a user
  DELETE /users/{user_id}       Delete a user

Tasks:
  POST   /users/{user_id}/tasks       Create a task for user
  GET    /users/{user_id}/tasks       List user's tasks (filtered)
  GET    /tasks/{task_id}             Get a single task
  PATCH  /tasks/{task_id}             Update a task
  DELETE /tasks/{task_id}             Delete a task
  POST   /tasks/{task_id}/complete    Mark task as complete

Model Definitions

# models/user.py
from pydantic import BaseModel, Field, EmailStr, ConfigDict
from typing import Optional, List
from datetime import datetime
from enum import Enum

class UserBase(BaseModel):
    """Base user fields shared across schemas."""
    email: EmailStr = Field(
        ...,
        description="User's email address (unique)",
        examples=["user@example.com"]
    )
    name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="User's display name",
        examples=["John Doe"]
    )

class UserCreate(UserBase):
    """Schema for creating a new user."""
    password: str = Field(
        ...,
        min_length=8,
        max_length=100,
        description="Password (min 8 characters)",
        examples=["securepassword123"]
    )

class UserUpdate(BaseModel):
    """Schema for updating a user. All fields optional."""
    email: Optional[EmailStr] = Field(
        None,
        description="New email address"
    )
    name: Optional[str] = Field(
        None,
        min_length=1,
        max_length=100,
        description="New display name"
    )
    password: Optional[str] = Field(
        None,
        min_length=8,
        max_length=100,
        description="New password"
    )

class UserResponse(UserBase):
    """Schema for user responses."""
    id: int = Field(..., description="Unique user ID")
    created_at: datetime = Field(..., description="Account creation timestamp")
    task_count: int = Field(0, description="Number of tasks")

    model_config = ConfigDict(from_attributes=True)

class UserListResponse(BaseModel):
    """Paginated list of users."""
    items: List[UserResponse]
    total: int = Field(..., description="Total number of users")
    page: int = Field(..., ge=1, description="Current page")
    per_page: int = Field(..., ge=1, le=100, description="Items per page")
    pages: int = Field(..., ge=0, description="Total pages")


# models/task.py
class TaskStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    completed = "completed"
    cancelled = "cancelled"

class TaskPriority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    urgent = "urgent"

class TaskBase(BaseModel):
    """Base task fields."""
    title: str = Field(
        ...,
        min_length=1,
        max_length=200,
        description="Task title",
        examples=["Complete project documentation"]
    )
    description: Optional[str] = Field(
        None,
        max_length=2000,
        description="Detailed task description"
    )
    priority: TaskPriority = Field(
        TaskPriority.medium,
        description="Task priority level"
    )
    due_date: Optional[datetime] = Field(
        None,
        description="Task due date"
    )

class TaskCreate(TaskBase):
    """Schema for creating a task."""
    pass

class TaskUpdate(BaseModel):
    """Schema for updating a task. All fields optional."""
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=2000)
    priority: Optional[TaskPriority] = None
    status: Optional[TaskStatus] = None
    due_date: Optional[datetime] = None

class TaskResponse(TaskBase):
    """Schema for task responses."""
    id: int
    user_id: int
    status: TaskStatus
    created_at: datetime
    updated_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None

    model_config = ConfigDict(from_attributes=True)

class TaskListResponse(BaseModel):
    """Paginated/filtered list of tasks."""
    items: List[TaskResponse]
    total: int
    page: int
    per_page: int
    pages: int
    filters_applied: dict = Field(
        default_factory=dict,
        description="Active filters"
    )

Query Parameter Models

# models/queries.py
from pydantic import BaseModel, Field
from typing import Optional, List

class PaginationParams(BaseModel):
    """Reusable pagination parameters."""
    page: int = Field(1, ge=1, description="Page number")
    per_page: int = Field(10, ge=1, le=100, description="Items per page")

class TaskFilterParams(BaseModel):
    """Task filtering parameters."""
    status: Optional[TaskStatus] = Field(None, description="Filter by status")
    priority: Optional[TaskPriority] = Field(None, description="Filter by priority")
    overdue: Optional[bool] = Field(None, description="Only overdue tasks")
    search: Optional[str] = Field(None, description="Search in title/description")

Error Response Models

# models/errors.py
from pydantic import BaseModel, Field
from typing import List, Optional, Any

class ValidationErrorDetail(BaseModel):
    """Single validation error."""
    field: str = Field(..., description="Field path (e.g., 'body.email')")
    message: str = Field(..., description="Error message")
    type: str = Field(..., description="Error type code")
    input: Optional[Any] = Field(None, description="Invalid input value")

class ValidationErrorResponse(BaseModel):
    """Validation error response (422)."""
    detail: str = "Validation failed"
    errors: List[ValidationErrorDetail]

class NotFoundResponse(BaseModel):
    """Not found error response (404)."""
    detail: str = Field(..., examples=["User not found"])

class ConflictResponse(BaseModel):
    """Conflict error response (409)."""
    detail: str = Field(..., examples=["Email already registered"])

---

Solution Architecture

Project Structure

task-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI application
│   ├── dependencies.py      # Shared dependencies
│   │
│   ├── models/              # Pydantic models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── task.py
│   │   ├── queries.py
│   │   └── errors.py
│   │
│   ├── routes/              # API routes
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── tasks.py
│   │
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── task_service.py
│   │
│   └── db/                  # Database (can use in-memory for learning)
│       ├── __init__.py
│       └── repository.py
│
├── tests/
│   ├── __init__.py
│   ├── test_users.py
│   └── test_tasks.py
│
├── pyproject.toml
└── README.md

Component Diagram

┌─────────────────────────────────────────────────────────────────┐
│                         main.py                                  │
│  - FastAPI app instance                                         │
│  - Exception handlers                                           │
│  - Route registration                                           │
│  - Middleware                                                    │
└─────────────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐
│   routes/users.py │ │  routes/tasks.py  │ │   dependencies.py │
│  - @router        │ │  - @router        │ │  - get_db         │
│  - CRUD endpoints │ │  - CRUD endpoints │ │  - get_user       │
│  - Query params   │ │  - Query params   │ │  - pagination     │
└─────────┬─────────┘ └─────────┬─────────┘ └───────────────────┘
          │                     │
          ▼                     ▼
┌─────────────────────────────────────────────────────────────────┐
│                     models/                                      │
│  - UserCreate, UserUpdate, UserResponse                         │
│  - TaskCreate, TaskUpdate, TaskResponse                         │
│  - PaginationParams, TaskFilterParams                           │
│  - Error models                                                 │
└─────────────────────────────────────────────────────────────────┘

Request/Response Flow

HTTP Request
     │
     ▼
┌─────────────────┐
│  FastAPI Route  │──► Path/Query/Header Validation
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Request Model   │──► Body Validation (Pydantic)
│ (UserCreate)    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Service Layer   │──► Business Logic
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Database/Repo   │──► Data Persistence
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Response Model  │──► Response Serialization
│ (UserResponse)  │    (Fields filtered, types converted)
└────────┬────────┘
         │
         ▼
HTTP Response (JSON)

---

Phased Implementation Guide

Phase 1: Project Setup (30 minutes)

Goal: Create a running FastAPI application.

1. Create project structure
2. Install dependencies:

   pip install fastapi uvicorn pydantic email-validator
   

3. Create basic main.py:

   from fastapi import FastAPI
   
   app = FastAPI(
       title="Task Management API",
       description="A RESTful API for managing users and tasks",
       version="1.0.0"
   )
   
   @app.get("/health")
   async def health_check():
       return {"status": "healthy"}
   

4. Run with: uvicorn app.main:app --reload

Checkpoint: Visit http://localhost:8000/docs to see Swagger UI.

Phase 2: User Models (1 hour)

Goal: Create all user-related Pydantic models.

1. Create models/user.py with:
   • UserBase, UserCreate, UserUpdate
   • UserResponse with from_attributes
   • UserListResponse for pagination
2. Add Field metadata for documentation
3. Add examples for OpenAPI

Checkpoint: Models import without errors.

Phase 3: User CRUD Endpoints (2 hours)

Goal: Implement user CRUD with validation.

1. Create routes/users.py:

   from fastapi import APIRouter, HTTPException, Query
   
   router = APIRouter(prefix="/users", tags=["users"])
   
   @router.post("", response_model=UserResponse, status_code=201)
   async def create_user(user: UserCreate):
       ...
   
   @router.get("", response_model=UserListResponse)
   async def list_users(
       page: int = Query(1, ge=1),
       per_page: int = Query(10, ge=1, le=100)
   ):
       ...
   

2. Implement in-memory storage or simple database
3. Handle email uniqueness constraint

Checkpoint: Can create, read, update, delete users via Swagger.

Phase 4: Custom Error Handling (1 hour)

Goal: Beautiful, consistent error responses.

1. Create custom exception handler:

   @app.exception_handler(RequestValidationError)
   async def validation_exception_handler(request, exc):
       ...
   

2. Create HTTPException handler for 404, 409, etc.
3. Document error responses in routes:

   @router.get(
       "/{user_id}",
       response_model=UserResponse,
       responses={404: {"model": NotFoundResponse}}
   )
   

Checkpoint: Errors return consistent JSON structure.

Phase 5: Task Models and Endpoints (2 hours)

Goal: Add task management with filtering.

1. Create models/task.py with all schemas
2. Create routes/tasks.py with CRUD
3. Implement nested route for user’s tasks
4. Add filtering with query parameters

Checkpoint: Full task CRUD works with filtering.

Phase 6: Advanced Features (2 hours)

Goal: Production-ready touches.

1. Add pagination helper dependency
2. Implement search in tasks
3. Add request ID header
4. Add rate limiting (optional)
5. Complete OpenAPI documentation

Checkpoint: API is fully functional and well-documented.

---

Testing Strategy

Unit Tests for Models

# tests/test_models.py
import pytest
from pydantic import ValidationError
from app.models.user import UserCreate, UserUpdate, UserResponse

def test_user_create_valid():
    user = UserCreate(
        email="test@example.com",
        name="Test User",
        password="securepassword123"
    )
    assert user.email == "test@example.com"

def test_user_create_invalid_email():
    with pytest.raises(ValidationError) as exc:
        UserCreate(
            email="not-an-email",
            name="Test",
            password="password123"
        )
    assert "email" in str(exc.value)

def test_user_create_short_password():
    with pytest.raises(ValidationError) as exc:
        UserCreate(
            email="test@example.com",
            name="Test",
            password="short"
        )
    errors = exc.value.errors()
    assert any(e["loc"] == ("password",) for e in errors)

def test_user_update_all_optional():
    # All fields None is valid for PATCH
    update = UserUpdate()
    assert update.email is None
    assert update.name is None

def test_user_response_from_attributes():
    class MockUser:
        id = 1
        email = "test@example.com"
        name = "Test"
        created_at = datetime.now()
        task_count = 5

    response = UserResponse.model_validate(MockUser())
    assert response.id == 1
    assert response.task_count == 5

API Integration Tests

# tests/test_users_api.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_user_success():
    response = client.post("/users", json={
        "email": "new@example.com",
        "name": "New User",
        "password": "securepassword123"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "new@example.com"
    assert "id" in data
    assert "password" not in data  # Never returned!

def test_create_user_validation_error():
    response = client.post("/users", json={
        "email": "not-an-email",
        "name": "",
        "password": "short"
    })
    assert response.status_code == 422
    data = response.json()
    assert "errors" in data
    assert len(data["errors"]) >= 3  # email, name, password

def test_create_user_duplicate_email():
    # First user
    client.post("/users", json={
        "email": "duplicate@example.com",
        "name": "First",
        "password": "password123"
    })
    # Second with same email
    response = client.post("/users", json={
        "email": "duplicate@example.com",
        "name": "Second",
        "password": "password123"
    })
    assert response.status_code == 409

def test_list_users_pagination():
    response = client.get("/users?page=1&per_page=5")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data
    assert data["page"] == 1
    assert data["per_page"] == 5

def test_get_user_not_found():
    response = client.get("/users/99999")
    assert response.status_code == 404

def test_update_user_partial():
    # Create user
    create_resp = client.post("/users", json={
        "email": "update@example.com",
        "name": "Original Name",
        "password": "password123"
    })
    user_id = create_resp.json()["id"]

    # Update only name
    response = client.patch(f"/users/{user_id}", json={
        "name": "Updated Name"
    })
    assert response.status_code == 200
    assert response.json()["name"] == "Updated Name"
    assert response.json()["email"] == "update@example.com"  # Unchanged

Task API Tests

# tests/test_tasks_api.py

def test_create_task_for_user():
    # Create user first
    user_resp = client.post("/users", json={...})
    user_id = user_resp.json()["id"]

    response = client.post(f"/users/{user_id}/tasks", json={
        "title": "Complete project",
        "priority": "high"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Complete project"
    assert data["status"] == "pending"  # Default

def test_filter_tasks_by_status():
    response = client.get(f"/users/{user_id}/tasks?status=pending")
    assert response.status_code == 200
    for task in response.json()["items"]:
        assert task["status"] == "pending"

def test_task_priority_enum_validation():
    response = client.post(f"/users/{user_id}/tasks", json={
        "title": "Test",
        "priority": "invalid"  # Not a valid priority
    })
    assert response.status_code == 422

---

Common Pitfalls and Debugging

Pitfall 1: Missing from_attributes

Problem: Returning ORM objects fails with validation error.

Symptom:

pydantic_core._pydantic_core.ValidationError: 1 validation error for UserResponse
  Input should be a valid dictionary or instance of UserResponse

Solution: Add model_config = ConfigDict(from_attributes=True)

Pitfall 2: Response Model Filtering Unexpected Fields

Problem: Extra fields in response are silently dropped.

Solution: This is intentional! Response models filter to declared fields only. If you need dynamic fields, use dict or customize serialization.

Pitfall 3: Optional vs None Default

Problem: Optional[str] still requires the field.

# This still requires 'name' to be present (can be null)
name: Optional[str]

# This makes 'name' optional with default None
name: Optional[str] = None

# Or in Python 3.10+
name: str | None = None

Pitfall 4: Circular Imports

Problem: UserResponse references TaskResponse which references UserResponse.

Solution: Use forward references:

from __future__ import annotations

class UserResponse(BaseModel):
    tasks: list[TaskResponse]  # Forward reference as string

# Or use TYPE_CHECKING
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from .task import TaskResponse

Pitfall 5: Path vs Query Parameter Confusion

Problem: /users/{user_id} getting query parameter instead.

# Path parameter
@router.get("/{user_id}")
async def get_user(user_id: int):  # From path

# Query parameter
@router.get("/")
async def list_users(user_id: int = Query(...)):  # From query string

Pitfall 6: Validation Error Not Showing Input

Problem: Can’t see what invalid value was submitted.

Solution: Access error.get("input") in exception handler:

for error in exc.errors():
    print(f"Field {error['loc']}: got {error.get('input')}")

---

Extensions and Challenges

Extension 1: API Versioning

Support multiple API versions:

app_v1 = FastAPI(prefix="/api/v1")
app_v2 = FastAPI(prefix="/api/v2")

# V1 models
class UserResponseV1(BaseModel):
    id: int
    name: str

# V2 models with breaking changes
class UserResponseV2(BaseModel):
    id: int
    full_name: str  # Renamed field
    email_verified: bool  # New field

Extension 2: Bulk Operations

Create multiple resources at once:

@router.post("/bulk", response_model=BulkCreateResponse)
async def create_users_bulk(users: List[UserCreate]):
    results = []
    for user in users:
        try:
            created = create_user(user)
            results.append({"status": "created", "id": created.id})
        except Exception as e:
            results.append({"status": "error", "message": str(e)})
    return {"results": results}

Extension 3: Field-Level Permissions

Different fields visible based on user role:

def get_response_model(user_role: str):
    if user_role == "admin":
        return UserAdminResponse  # Includes sensitive fields
    return UserPublicResponse  # Limited fields

Extension 4: Request Correlation

Track requests across services:

from uuid import uuid4

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid4()))
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

Extension 5: Rate Limiting with Pydantic

Define rate limit rules with Pydantic:

class RateLimitRule(BaseModel):
    endpoint: str
    requests_per_minute: int = Field(ge=1)
    burst: int = Field(ge=1)

rate_limits = [
    RateLimitRule(endpoint="/users", requests_per_minute=60),
    RateLimitRule(endpoint="/tasks", requests_per_minute=120),
]

---

Real-World Connections

Where This Pattern Appears

1. FastAPI Applications - Default pattern for all FastAPI apps
2. Django REST Framework - Serializers are similar to Pydantic models
3. GraphQL APIs - Type definitions serve similar purpose
4. gRPC Services - Protocol Buffers define message types

Industry Examples

• Stripe API - Exemplary API documentation and error handling
• GitHub API - Pagination patterns
• Twilio - Request/response model separation
• Anthropic/OpenAI APIs - Structured request/response models

Production Considerations

1. Rate Limiting - Protect your API from abuse
2. Authentication - Validate API keys/tokens
3. CORS - Configure for frontend access
4. Logging - Log requests/responses for debugging
5. Monitoring - Track response times and error rates

---

Self-Assessment Checklist

Core Understanding

• Can I explain why input and output models should be separate?
• Can I describe the request validation pipeline in FastAPI?
• Can I explain how response_model filters the response?
• Can I describe when to use Query, Path, Body, Header?

Implementation Skills

• Can I create a full CRUD API with proper models?
• Can I implement custom error handlers?
• Can I add pagination to list endpoints?
• Can I implement query parameter filtering?

Documentation

• Can I add descriptions and examples to fields?
• Can I document error responses in OpenAPI?
• Can I generate comprehensive API documentation?

Mastery Indicators

• API handles all edge cases gracefully
• Error messages are helpful for API consumers
• OpenAPI documentation is complete and accurate
• Tests cover both success and error cases

---

Resources

Documentation

• FastAPI Documentation
• Pydantic with FastAPI
• Response Model
• Request Validation

Books

• “Building Data Science Applications with FastAPI” by François Voron

Related Projects

• FastAPI Full Stack Template
• FastAPI Best Practices