API Request and Response Design

Master the art of designing clean, intuitive request and response structures that scale

Designing Developer-Friendly APIs

The structure of your API's requests and responses determines how intuitive and maintainable your API will be. Learn pagination strategies, filtering patterns, response envelope designs, and content negotiation that make your APIs a joy to consume. This lesson covers the patterns used by top APIs like GitHub, Stripe, and Twitter.

Request Body Design and Validation

A well-designed request body is predictable, validates thoroughly, and provides clear error messages. Pydantic models make this easy in FastAPI by combining type hints with validation.

Basic Request Validation

from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Optional
from datetime import datetime

class CreateUserRequest(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr  # Validates email format
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=13, le=120)
    bio: Optional[str] = Field(None, max_length=500)

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v.lower()  # Normalize to lowercase

    @field_validator('password')
    @classmethod
    def password_strength(cls, v):
        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 digit')
        return v

@app.post("/users", status_code=201)
async def create_user(user: CreateUserRequest):
    # Pydantic has already validated all fields
    # Create user in database
    new_user = await db.create_user(user.model_dump())
    return {"id": new_user.id, "username": user.username}

Result: Pydantic automatically validates all fields before your function runs. If validation fails, FastAPI returns a detailed 422 error with field-specific messages.

Nested Models and Complex Validation

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    state: str = Field(..., min_length=2, max_length=2)
    zip_code: str = Field(..., pattern=r'^\d{5}$')

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(..., gt=0, le=100)
    price: float = Field(..., gt=0)

class CreateOrderRequest(BaseModel):
    customer_id: int
    items: List[OrderItem] = Field(..., min_length=1, max_length=50)
    shipping_address: Address
    billing_address: Optional[Address] = None
    promo_code: Optional[str] = None

    @model_validator(mode='after')
    def set_billing_address(self):
        # Default billing address to shipping if not provided
        if not self.billing_address:
            self.billing_address = self.shipping_address
        return self

    @field_validator('items')
    @classmethod
    def validate_total_quantity(cls, items):
        total_qty = sum(item.quantity for item in items)
        if total_qty > 100:
            raise ValueError('Total quantity cannot exceed 100 items')
        return items

@app.post("/orders")
async def create_order(order: CreateOrderRequest):
    total = sum(item.price * item.quantity for item in order.items)
    # Process order...
    return {"order_id": 12345, "total": total}

Result: Nested models validate complex hierarchical data. Custom validators enforce business rules like maximum quantities or address defaults.

Bad: No Validation

@app.post("/users")
async def create_user(data: dict):
    # No validation!
    username = data.get("username")
    email = data.get("email")
    # Anything could happen...

Good: Validated Model

@app.post("/users")
async def create_user(
    user: CreateUserRequest
):
    # Guaranteed valid data
    # Type-safe access

Query Parameters vs Path Parameters vs Request Body

Choosing the right parameter location makes your API intuitive and RESTful. Each location has specific use cases and conventions.

Parameter Type	Use Case	Example	Characteristics
Path Parameters	Identify a specific resource	`/users/123`	Required, part of URL structure
Query Parameters	Filter, sort, paginate collections	`/users?role=admin&page=2`	Optional, key-value pairs
Request Body	Send complex/large data structures	JSON payload in POST/PUT	Structured data, not in URL
Headers	Authentication, content type, metadata	`Authorization: Bearer token`	Not specific to resource

Decision Guide

from fastapi import Query, Path, Body
from typing import Optional

# PATH PARAMETERS: Resource identification (required)
@app.get("/users/{user_id}/posts/{post_id}")
async def get_post(
    user_id: int = Path(..., gt=0),
    post_id: int = Path(..., gt=0)
):
    return {"user_id": user_id, "post_id": post_id}

# QUERY PARAMETERS: Optional filtering/sorting
@app.get("/products")
async def list_products(
    category: Optional[str] = Query(None, max_length=50),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    sort: str = Query("name", pattern="^(name|price|created_at)$"),
    order: str = Query("asc", pattern="^(asc|desc)$"),
    page: int = Query(1, ge=1),
    limit: int = Query(20, ge=1, le=100)
):
    # Filter and paginate products
    return {"products": [], "page": page, "total": 0}

# REQUEST BODY: Complex data submission
class UpdateProfileRequest(BaseModel):
    display_name: Optional[str] = None
    bio: Optional[str] = None
    avatar_url: Optional[str] = None
    social_links: Optional[dict] = None
    preferences: Optional[dict] = None

@app.patch("/users/{user_id}/profile")
async def update_profile(
    user_id: int = Path(..., gt=0),
    profile: UpdateProfileRequest = Body(...),
    notify_followers: bool = Query(False)  # Behavior modifier
):
    # user_id: path (which resource)
    # profile: body (what to change)
    # notify_followers: query (how to behave)
    return {"user_id": user_id, "updated": profile.model_dump(exclude_none=True)}

Result: Path identifies the resource, query modifies behavior, body contains the data. This separation makes the API predictable.

Common Mistakes

Using query params for resource IDs: Use /users/123 not /users?id=123
Required query params: If it's required, it should probably be a path param
Sending data in GET request bodies: Use query params for filtering GET requests
Putting large data in query strings: URLs have length limits; use request body

Response Design Patterns

How you structure responses affects client code complexity. The choice between wrapped and unwrapped responses is one of the most important API design decisions.

Unwrapped Responses (Recommended for REST)

Unwrapped responses return the resource data directly without any envelope. For a single resource, you return the object itself; for collections, you return an array. This is the simplest and most RESTful approach, HTTP headers carry metadata like status codes, content type, and caching directives, while the body contains only the actual data.

When to use: Most REST APIs, especially when following standards. Popular with APIs like GitHub, Stripe, and Twitter. Ideal when HTTP headers are sufficient for metadata.

from pydantic import BaseModel
from typing import List

class User(BaseModel):
    id: int
    username: str
    email: str

# Single resource - return object directly
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    user = await db.get_user(user_id)
    return user  # FastAPI serializes to JSON

# Response:
# {
#   "id": 123,
#   "username": "john_doe",
#   "email": "john@example.com"
# }

# Collection - return array directly
@app.get("/users", response_model=List[User])
async def list_users():
    users = await db.get_all_users()
    return users

# Response:
# [
#   {"id": 1, "username": "alice", "email": "alice@example.com"},
#   {"id": 2, "username": "bob", "email": "bob@example.com"}
# ]

Result: Clean, minimal responses. HTTP headers provide metadata (status, content-type). This is the REST convention used by GitHub, Stripe, and most modern APIs.

Wrapped Responses (When You Need Metadata)

Wrapped responses enclose the actual data inside a consistent envelope structure, typically with a data field containing the resource(s) and additional fields for pagination, errors, or other metadata. While this adds structure, it also adds verbosity and makes client code slightly more complex since clients must unwrap the data field.

When to use: When you need rich pagination metadata (page numbers, totals), consistent error formats across all endpoints, or when building a GraphQL-style API. Also useful when HTTP headers alone cannot convey all necessary metadata.

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

T = TypeVar('T')

class PaginatedResponse(BaseModel, Generic[T]):
    data: List[T]
    page: int
    per_page: int
    total: int
    total_pages: int

class ErrorDetail(BaseModel):
    field: str
    message: str

class ErrorResponse(BaseModel):
    error: str
    details: Optional[List[ErrorDetail]] = None

# Paginated collection response
@app.get("/users")
async def list_users(page: int = 1, per_page: int = 20):
    users = await db.get_users(page=page, per_page=per_page)
    total = await db.count_users()

    return PaginatedResponse(
        data=users,
        page=page,
        per_page=per_page,
        total=total,
        total_pages=(total + per_page - 1) // per_page
    )

# Response:
# {
#   "data": [
#     {"id": 1, "username": "alice"},
#     {"id": 2, "username": "bob"}
#   ],
#   "page": 1,
#   "per_page": 20,
#   "total": 150,
#   "total_pages": 8
# }

# Error response
@app.exception_handler(ValidationException)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="Validation failed",
            details=[
                ErrorDetail(field="email", message="Invalid email format"),
                ErrorDetail(field="age", message="Must be at least 13")
            ]
        ).model_dump()
    )

Result: Wrapped responses include metadata alongside data. Use for pagination info, error details, or when you need consistent response structure across all endpoints.

Unwrapped Pros

Simpler client code
RESTful convention
Easier to cache
Uses HTTP headers for metadata

Wrapped Pros

Consistent structure
Metadata in response body
Better for pagination
Easier to version

Bonus: Automated Response Standardization

If you choose the wrapped response strategy, manually wrapping every endpoint can be tedious and error-prone. The core-apis library provides a decorator that automatically standardizes all your API responses with consistent formatting, status codes, and error handling.

What it does:

Wraps responses in consistent structure
Handles errors uniformly across endpoints
Adds metadata automatically (status, code, error)
Reduces boilerplate code
Ensures API consistency

# Install
pip install core-apis

# Usage
from core_apis.decorators import response_wrapper

@response_wrapper
def get_user(user_id: int):
    return {...}

# Automatically wraps to:
# {
#   "code": ...,
#   "status": "success",
#   "result": {"id": 1, "name": "John"},
#   "error": null
# }

PyPI Package Documentation

Note: This is particularly useful for large teams or microservices architectures where consistency across multiple APIs is crucial. The decorator approach keeps your endpoint code clean while ensuring uniform response structures.

Pagination Strategies

Pagination prevents large datasets from overwhelming clients and servers. Different strategies suit different use cases.

1. Offset-Based Pagination (Simplest)

from fastapi import Query
from typing import List

@app.get("/users")
async def list_users(
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page")
):
    offset = (page - 1) * per_page

    # SQL: SELECT * FROM users LIMIT per_page OFFSET offset
    users = await db.get_users(limit=per_page, offset=offset)
    total = await db.count_users()

    return {
        "data": users,
        "pagination": {
            "page": page,
            "per_page": per_page,
            "total": total,
            "total_pages": (total + per_page - 1) // per_page,
            "has_next": page * per_page < total,
            "has_prev": page > 1
        }
    }

# Request: GET /users?page=3&per_page=20
# Response:
# {
#   "data": [...],
#   "pagination": {
#     "page": 3,
#     "per_page": 20,
#     "total": 150,
#     "total_pages": 8,
#     "has_next": true,
#     "has_prev": true
#   }
# }

Result: Simple to implement and understand. Works well for small-to-medium datasets.
Downside: Performance degrades with large offsets (page 1000). Items can be skipped/duplicated if data changes between requests.

2. Cursor-Based Pagination (Recommended for Feeds)

from typing import Optional
import base64
import json

def encode_cursor(value: int) -> str:
    """Encode cursor (usually last item's ID or timestamp)"""
    return base64.b64encode(json.dumps({"id": value}).encode()).decode()

def decode_cursor(cursor: str) -> int:
    """Decode cursor to get last seen ID"""
    return json.loads(base64.b64decode(cursor))["id"]

@app.get("/posts")
async def list_posts(
    cursor: Optional[str] = Query(None, description="Pagination cursor"),
    limit: int = Query(20, ge=1, le=100)
):
    if cursor:
        last_id = decode_cursor(cursor)
        # Get posts after this ID
        posts = await db.query(
            "SELECT * FROM posts WHERE id > $1 ORDER BY id LIMIT $2",
            last_id, limit
        )
    else:
        # First page
        posts = await db.query(
            "SELECT * FROM posts ORDER BY id LIMIT $1",
            limit
        )

    next_cursor = None
    if len(posts) == limit:
        # More results exist
        next_cursor = encode_cursor(posts[-1]["id"])

    return {
        "data": posts,
        "pagination": {
            "next_cursor": next_cursor,
            "has_more": next_cursor is not None
        }
    }

# Request 1: GET /posts?limit=20
# Response: { "data": [...], "pagination": { "next_cursor": "eyJpZCI6MjB9" } }
#
# Request 2: GET /posts?cursor=eyJpZCI6MjB9&limit=20
# Response: { "data": [...], "pagination": { "next_cursor": "eyJpZCI6NDB9" } }

Result: Consistent results even if data changes. Efficient for large datasets.
Use case: Social media feeds, real-time data, infinite scroll.
Downside: Can't jump to arbitrary pages.

3. Keyset Pagination (Best Performance)

from datetime import datetime
from typing import Optional

@app.get("/articles")
async def list_articles(
    created_before: Optional[datetime] = Query(None),
    limit: int = Query(20, ge=1, le=100)
):
    if created_before:
        # Get articles created before this timestamp
        articles = await db.query("""
            SELECT * FROM articles
            WHERE created_at < $1
            ORDER BY created_at DESC
            LIMIT $2
        """, created_before, limit)
    else:
        articles = await db.query("""
            SELECT * FROM articles
            ORDER BY created_at DESC
            LIMIT $1
        """, limit)

    next_url = None
    if len(articles) == limit:
        last_created = articles[-1]["created_at"]
        next_url = f"/articles?created_before={last_created.isoformat()}&limit={limit}"

    return {
        "data": articles,
        "pagination": {
            "next": next_url
        }
    }

# Request 1: GET /articles?limit=20
# Response: {
#   "data": [...],
#   "pagination": {
#     "next": "/articles?created_before=2024-01-15T10:30:00Z&limit=20"
#   }
# }

Result: Extremely efficient - uses indexed columns. No offset calculation.
Use case: Time-series data, logs, any naturally ordered dataset.
Requirement: Must have an indexed, unique, sequential column.

Strategy	Pros	Cons	Best For
Offset-Based	Simple, jump to any page	Slow at high offsets, inconsistent	Admin panels, small datasets
Cursor-Based	Consistent, efficient	Can't jump to arbitrary page	Feeds, infinite scroll
Keyset	Fastest, most consistent	Requires indexed column	Time-series, logs

Filtering, Sorting, and Field Selection

Let clients customize what data they receive and how it's organized. These patterns reduce bandwidth and improve performance.

Filtering Collections

from fastapi import Query
from typing import Optional, List
from datetime import datetime
from enum import Enum

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class UserStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    SUSPENDED = "suspended"

@app.get("/users")
async def list_users(
    # Exact match filters
    role: Optional[UserRole] = Query(None),
    status: Optional[UserStatus] = Query(None),

    # Range filters
    created_after: Optional[datetime] = Query(None),
    created_before: Optional[datetime] = Query(None),
    min_age: Optional[int] = Query(None, ge=0),
    max_age: Optional[int] = Query(None, le=120),

    # Text search
    search: Optional[str] = Query(None, min_length=3, max_length=100),

    # Multiple values (array)
    country: Optional[List[str]] = Query(None),

    # Sorting
    sort: str = Query("created_at", pattern="^-?(username|email|created_at|age)$"),

    # Pagination
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100)
):
    # Build query dynamically
    query = "SELECT * FROM users WHERE 1=1"
    params = []

    if role:
        query += f" AND role = ${len(params) + 1}"
        params.append(role)

    if status:
        query += f" AND status = ${len(params) + 1}"
        params.append(status)

    if created_after:
        query += f" AND created_at >= ${len(params) + 1}"
        params.append(created_after)

    if search:
        query += f" AND (username ILIKE ${len(params) + 1} OR email ILIKE ${len(params) + 1})"
        params.append(f"%{search}%")

    if country:
        placeholders = ",".join(f"${i}" for i in range(len(params) + 1, len(params) + len(country) + 1))
        query += f" AND country IN ({placeholders})"
        params.extend(country)

    # Handle sorting (- prefix means DESC)
    sort_direction = "DESC" if sort.startswith("-") else "ASC"
    sort_field = sort.lstrip("-")
    query += f" ORDER BY {sort_field} {sort_direction}"

    # Pagination
    offset = (page - 1) * per_page
    query += f" LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}"
    params.extend([per_page, offset])

    users = await db.fetch_all(query, *params)

    return {"data": users, "page": page}

# Example queries:
# GET /users?role=admin&status=active
# GET /users?created_after=2024-01-01&sort=-created_at
# GET /users?search=john&country=US&country=CA
# GET /users?min_age=18&max_age=65&sort=age

Result: Flexible filtering that lets clients request exactly what they need. Use query parameters for filters, not custom endpoints for each combination.

Field Selection (Sparse Fieldsets)

from fastapi import Query
from typing import Optional, Set
from pydantic import BaseModel

class User(BaseModel):
    id: int
    username: str
    email: str
    bio: Optional[str]
    avatar_url: Optional[str]
    created_at: str
    updated_at: str
    followers_count: int
    following_count: int

    # Available fields can be retrieved via User.model_fields.keys()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    fields: Optional[str] = Query(
        None,
        description="Comma-separated list of fields to return",
        example="id,username,email"
    )
):
    user = await db.get_user(user_id)

    if fields:
        # Parse requested fields
        requested_fields = set(f.strip() for f in fields.split(','))

        # Validate fields exist
        valid_fields = set(User.model_fields.keys())
        invalid_fields = requested_fields - valid_fields
        if invalid_fields:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid fields: {', '.join(invalid_fields)}"
            )

        # Return only requested fields
        return {k: v for k, v in user.model_dump().items() if k in requested_fields}

    # Return all fields
    return user

# Request: GET /users/123
# Response: {
#   "id": 123,
#   "username": "john_doe",
#   "email": "john@example.com",
#   "bio": "Software engineer",
#   "avatar_url": "https://...",
#   "created_at": "2024-01-15T10:30:00Z",
#   "updated_at": "2024-01-20T14:20:00Z",
#   "followers_count": 1250,
#   "following_count": 340
# }

# Request: GET /users/123?fields=id,username,avatar_url
# Response: {
#   "id": 123,
#   "username": "john_doe",
#   "avatar_url": "https://..."
# }

Result: Clients request only the fields they need, reducing bandwidth and improving performance. Similar to GraphQL's field selection but in REST.

Query Parameter Naming Conventions

?sort=-created_at - Use minus prefix for descending order
?filter[status]=active - Bracket notation for grouped filters
?page[size]=20&page[number]=3 - JSON API style pagination
?fields=id,name,email - Comma-separated for field selection
?include=author,comments - Related resources to include

Partial Responses and Updates

Allow clients to update only specific fields without sending the entire resource. This is especially important for large objects.

PATCH with Partial Updates

from pydantic import BaseModel
from typing import Optional

class UserUpdate(BaseModel):
    # All fields optional for partial updates
    username: Optional[str] = None
    email: Optional[str] = None
    bio: Optional[str] = None
    avatar_url: Optional[str] = None

    class Config:
        # Only include fields that were explicitly set
        # (None means "don't update", missing means "don't update")
        # This distinguishes between "set to null" vs "don't change"
        pass

@app.patch("/users/{user_id}")
async def update_user(user_id: int, updates: UserUpdate):
    # Get current user
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404)

    # Apply only provided fields
    update_data = updates.model_dump(exclude_unset=True)

    # Validate business rules
    if 'email' in update_data:
        # Check if email already exists
        existing = await db.get_user_by_email(update_data['email'])
        if existing and existing.id != user_id:
            raise HTTPException(status_code=409, detail="Email already in use")

    # Update database
    if update_data:
        await db.update_user(user_id, update_data)

    # Return updated user
    updated_user = await db.get_user(user_id)
    return updated_user

# Request: PATCH /users/123
# Body: {"bio": "Updated bio", "avatar_url": "https://new-avatar.jpg"}
# Result: Only bio and avatar_url are updated, other fields unchanged

# Request: PATCH /users/123
# Body: {"email": "newemail@example.com"}
# Result: Only email is updated

Result: exclude_unset=True ensures only fields explicitly provided in the request are updated. This prevents accidentally clearing fields.

Nullable vs Optional Fields

from pydantic import BaseModel, Field
from typing import Optional

class UserProfileUpdate(BaseModel):
    # Optional[str] = Field(None) means:
    # - Not included in request: don't update
    # - Included with value: update to that value
    # - Included with null: set to null/empty

    display_name: Optional[str] = Field(None)
    bio: Optional[str] = Field(None)
    website: Optional[str] = Field(None)

@app.patch("/users/{user_id}/profile")
async def update_profile(user_id: int, profile: UserProfileUpdate):
    user = await db.get_user(user_id)

    # This correctly handles:
    # 1. Missing field - don't update
    # 2. null value - clear the field
    # 3. String value - update to new value

    update_data = {}

    # Check each field explicitly
    if profile.model_fields_set:  # Fields that were set in the request
        for field in profile.model_fields_set:
            value = getattr(profile, field)
            update_data[field] = value  # Can be None to clear

    if update_data:
        await db.update_user_profile(user_id, update_data)

    return await db.get_user(user_id)

# Example 1: Update only display_name
# PATCH /users/123/profile
# {"display_name": "John Doe"}
# Result: display_name updated, bio and website unchanged

# Example 2: Clear the bio
# PATCH /users/123/profile
# {"bio": null}
# Result: bio set to null, other fields unchanged

# Example 3: Update multiple fields
# PATCH /users/123/profile
# {"display_name": "Jane", "website": "https://jane.com"}
# Result: Both fields updated, bio unchanged

Result: Proper distinction between "not provided", "set to null", and "set to value". Use model_fields_set to detect which fields were in the request.

Content Negotiation

Content negotiation allows the same endpoint to return different representations of a resource based on client preferences (JSON, XML, CSV, etc.).

Accept Header-Based Negotiation

from fastapi import Request, Response
from fastapi.responses import JSONResponse, PlainTextResponse
import csv
import io

@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404)

    # Get Accept header
    accept = request.headers.get("accept", "application/json")

    # JSON response (default)
    # For simplicity using if/else statements, strategy pattern is
    # recommended for better maintainability.
    if "application/json" in accept:
        return JSONResponse(content=user)

    # XML response
    elif "application/xml" in accept or "text/xml" in accept:
        xml = f"""<?xml version="1.0"?>
<user>
    <id>{user['id']}</id>
    <username>{user['username']}</username>
    <email>{user['email']}</email>
</user>"""
        return Response(content=xml, media_type="application/xml")

    # CSV response
    elif "text/csv" in accept:
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=user.keys())
        writer.writeheader()
        writer.writerow(user)

        return Response(
            content=output.getvalue(),
            media_type="text/csv",
            headers={
                "Content-Disposition": f"attachment; filename=user_{user_id}.csv"
            }
        )

    # Plain text
    elif "text/plain" in accept:
        text = "\n".join(f"{k}: {v}" for k, v in user.items())
        return PlainTextResponse(content=text)

    # Unsupported format
    else:
        raise HTTPException(
            status_code=406,  # Not Acceptable
            detail="Supported formats: application/json, application/xml, text/csv, text/plain"
        )

# Request with JSON (default):
# GET /users/123
# Accept: application/json
# Response: {"id": 123, "username": "john", "email": "john@example.com"}

# Request with XML:
# GET /users/123
# Accept: application/xml
# Response: <?xml version="1.0"?><user><id>123</id>...

# Request with CSV:
# GET /users/123
# Accept: text/csv
# Response: id,username,email\n123,john,john@example.com

Result: Same endpoint serves multiple formats. Return 406 Not Acceptable if the client requests an unsupported format.

Format Parameter Alternative

from enum import Enum
from fastapi import Query

class ResponseFormat(str, Enum):
    JSON = "json"
    XML = "xml"
    CSV = "csv"

@app.get("/users/export")
async def export_users(
    format: ResponseFormat = Query(ResponseFormat.JSON)
):
    users = await db.get_all_users()

    if format == ResponseFormat.JSON:
        return JSONResponse(content={"users": users})

    elif format == ResponseFormat.XML:
        xml = "<?xml version=\"1.0\"?><users>"
        for user in users:
            xml += f"<user><id>{user['id']}</id><username>{user['username']}</username></user>"
        xml += "</users>"
        return Response(content=xml, media_type="application/xml")

    elif format == ResponseFormat.CSV:
        output = io.StringIO()
        if users:
            writer = csv.DictWriter(output, fieldnames=users[0].keys())
            writer.writeheader()
            writer.writerows(users)

        return Response(
            content=output.getvalue(),
            media_type="text/csv",
            headers={"Content-Disposition": "attachment; filename=users.csv"}
        )

# Request: GET /users/export?format=json
# Request: GET /users/export?format=xml
# Request: GET /users/export?format=csv

Result: Query parameter approach is simpler for clients (no custom headers needed). Use for export/download endpoints where format selection is explicit.

When to Use Content Negotiation

Use Accept header: When the same resource has multiple representations and you want true REST compliance. Better for APIs consumed by diverse clients.

Use query parameter: For explicit export/download features where users choose the format. Simpler for web applications and human users.

File Uploads and Multipart Requests

Handle file uploads securely and efficiently. Multipart requests combine files with structured data in a single request. This section covers the fundamentals to get you started. For production patterns including S3 uploads, virus scanning, chunked uploads, and advanced security, see Lesson 21.

Single File Upload

from fastapi import File, UploadFile, HTTPException
import aiofiles
import os
from pathlib import Path

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".pdf", ".txt"}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # Validate file extension
    file_ext = Path(file.filename).suffix.lower()
    if file_ext not in ALLOWED_EXTENSIONS:
        raise HTTPException(
            status_code=400,
            detail=f"File type {file_ext} not allowed. Allowed: {ALLOWED_EXTENSIONS}"
        )

    # Read file in chunks to check size
    total_size = 0
    chunks = []

    while chunk := await file.read(8192):  # 8KB chunks
        total_size += len(chunk)
        if total_size > MAX_FILE_SIZE:
            raise HTTPException(
                status_code=413,
                detail=f"File too large. Maximum size: {MAX_FILE_SIZE / 1024 / 1024}MB"
            )
        chunks.append(chunk)

    # Generate unique filename
    import uuid
    unique_filename = f"{uuid.uuid4()}{file_ext}"
    file_path = UPLOAD_DIR / unique_filename

    # Save file
    async with aiofiles.open(file_path, 'wb') as f:
        for chunk in chunks:
            await f.write(chunk)

    return {
        "filename": file.filename,
        "stored_as": unique_filename,
        "size": total_size,
        "content_type": file.content_type,
        "url": f"/files/{unique_filename}"
    }

# Client request:
# POST /upload
# Content-Type: multipart/form-data
# Body: (binary file data)

# Response:
# {
#   "filename": "document.pdf",
#   "stored_as": "a1b2c3d4-e5f6-7890-abcd-ef1234567890.pdf",
#   "size": 245678,
#   "content_type": "application/pdf",
#   "url": "/files/a1b2c3d4-e5f6-7890-abcd-ef1234567890.pdf"
# }

Result: Secure file upload with validation (type, size), unique naming to prevent collisions, and async I/O for better performance.

Multiple Files with Metadata

from fastapi import File, UploadFile, Form
from typing import List
from pydantic import BaseModel

class DocumentMetadata(BaseModel):
    title: str
    description: str
    category: str
    tags: List[str]

@app.post("/documents")
async def upload_documents(
    # Files
    files: List[UploadFile] = File(..., description="Multiple files"),

    # Structured data as form fields
    title: str = Form(...),
    description: str = Form(...),
    category: str = Form(...),
    tags: str = Form(..., description="Comma-separated tags"),

    # Or as JSON string that we parse
    # metadata: str = Form(...)  # JSON string
):
    if len(files) > 10:
        raise HTTPException(status_code=400, detail="Maximum 10 files per request")

    # Parse tags
    tag_list = [tag.strip() for tag in tags.split(",")]

    uploaded_files = []

    for file in files:
        # Validate each file
        file_ext = Path(file.filename).suffix.lower()
        if file_ext not in ALLOWED_EXTENSIONS:
            continue  # Skip invalid files (or raise error)

        # Generate unique filename
        unique_filename = f"{uuid.uuid4()}{file_ext}"
        file_path = UPLOAD_DIR / unique_filename

        # Save file
        async with aiofiles.open(file_path, 'wb') as f:
            while chunk := await file.read(8192):
                await f.write(chunk)

        # Save metadata to database
        doc = await db.create_document({
            "filename": file.filename,
            "stored_filename": unique_filename,
            "title": title,
            "description": description,
            "category": category,
            "tags": tag_list,
            "content_type": file.content_type,
            "size": file.file.tell()
        })

        uploaded_files.append({
            "id": doc.id,
            "filename": file.filename,
            "url": f"/files/{unique_filename}"
        })

    return {
        "uploaded": len(uploaded_files),
        "files": uploaded_files
    }

# Client request (multipart/form-data):
# POST /documents
# Content-Type: multipart/form-data; boundary=----WebKitFormBoundary
#
# ------WebKitFormBoundary
# Content-Disposition: form-data; name="title"
# Project Documents
# ------WebKitFormBoundary
# Content-Disposition: form-data; name="description"
# Q1 2024 reports
# ------WebKitFormBoundary
# Content-Disposition: form-data; name="files"; filename="report1.pdf"
# Content-Type: application/pdf
# (binary data)
# ------WebKitFormBoundary
# Content-Disposition: form-data; name="files"; filename="report2.pdf"
# Content-Type: application/pdf
# (binary data)

Result: Upload multiple files with structured metadata in a single request. Combine File() for files and Form() for text data.

Direct Binary Upload (Large Files)

from fastapi import Request, Header
from typing import Optional

@app.put("/files/{file_id}")
async def upload_large_file(
    file_id: str,
    request: Request,
    content_type: Optional[str] = Header(None),
    content_length: Optional[int] = Header(None)
):
    """Direct binary upload for large files (streaming)"""

    # Validate content length
    if content_length and content_length > 100 * 1024 * 1024:  # 100 MB
        raise HTTPException(status_code=413, detail="File too large")

    file_path = UPLOAD_DIR / f"{file_id}.bin"

    # Stream directly to disk without loading into memory
    bytes_written = 0

    async with aiofiles.open(file_path, 'wb') as f:
        async for chunk in request.stream():
            bytes_written += len(chunk)

            # Check size limit during streaming
            if bytes_written > 100 * 1024 * 1024:
                await f.close()
                os.remove(file_path)  # Delete partial file
                raise HTTPException(status_code=413, detail="File too large")

            await f.write(chunk)

    return {
        "file_id": file_id,
        "size": bytes_written,
        "content_type": content_type
    }

# Client request:
# PUT /files/abc123
# Content-Type: application/octet-stream
# Content-Length: 52428800
# Body: (raw binary data)

# Use case: Large file uploads where multipart overhead is problematic
# e.g., video uploads, database backups, large datasets

Result: Stream large files directly to disk without loading into memory. Use for video uploads, backups, or any large binary data.

File Upload Security Checklist

Validate file types: Check extensions AND MIME types (can be spoofed)
Limit file size: Prevent DoS attacks with reasonable limits
Use unique filenames: Prevent path traversal attacks (../../etc/passwd)
Scan for malware: Use antivirus for production systems
Store outside web root: Prevent direct access to uploaded files
Sanitize filenames: Remove special characters, limit length
Rate limit uploads: Prevent abuse

Ready for Production File Handling?

These patterns work for simple applications. For production systems, you'll need advanced patterns: multi-layer validation (magic bytes), S3 direct uploads, virus scanning (ClamAV), chunked/resumable uploads, range requests, and background job integration. See Lesson 21: File Upload & Download Patterns for comprehensive production patterns used by Dropbox, Google Drive, and AWS.

Key Takeaways

Request Validation: Use Pydantic models for automatic validation with clear error messages. Custom validators enforce business rules.
Parameter Placement: Path for resource IDs, query for filters/options, body for complex data. This separation makes APIs intuitive.
Response Structure: Unwrapped responses are simpler and more RESTful. Wrap when you need pagination metadata or consistent error formats.
Pagination: Offset is simplest, cursor prevents inconsistencies, keyset is fastest. Choose based on your data access patterns.
Filtering & Sorting: Let clients customize responses with query parameters. Use consistent naming (sort=-field for DESC, fields=a,b,c for selection).
Partial Updates: PATCH with exclude_unset=True allows updating only specified fields without affecting others.
Content Negotiation: Same endpoint, multiple formats. Use Accept header for REST compliance or query parameter for simplicity.
File Uploads: Validate types and sizes, use unique filenames, stream large files to prevent memory issues. Security is critical. (Lesson 21 covers production patterns: S3, virus scanning, chunked uploads, and advanced security.)

Back to "ASGI vs WSGI"

Web APIsLesson 4

Next: "Authentication & Authorization"