FastAPI provides one of the most intuitive and powerful routing systems in any Python web framework. Built on top of Starlette and enhanced with Python type hints, FastAPI’s routing lets you define endpoints that automatically validate inputs, generate documentation, and handle serialization — all with minimal boilerplate. In this tutorial, we will explore every aspect of routing and request handling in FastAPI, from basic path operations to advanced patterns like dependency injection and API routers.

pip install fastapi uvicorn python-multipart aiofiles

1. Route Basics

In FastAPI, a route (also called a path operation) is the combination of an HTTP method and a URL path that maps to a Python function. You define routes using decorators on your FastAPI application instance.

1.1 The Application Instance

Every FastAPI application starts with creating an instance of the FastAPI class:

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="A demonstration of FastAPI routing",
    version="1.0.0"
)

1.2 Path Operation Decorators

FastAPI provides decorators for every standard HTTP method. Each decorator takes a path string and registers the decorated function as the handler for that path and method combination:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    """Handle GET requests to the root path."""
    return {"message": "Hello, World!"}

@app.post("/items")
def create_item():
    """Handle POST requests to create an item."""
    return {"message": "Item created"}

@app.put("/items/{item_id}")
def update_item(item_id: int):
    """Handle PUT requests to update an item."""
    return {"message": f"Item {item_id} updated"}

@app.patch("/items/{item_id}")
def partial_update_item(item_id: int):
    """Handle PATCH requests to partially update an item."""
    return {"message": f"Item {item_id} partially updated"}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    """Handle DELETE requests to delete an item."""
    return {"message": f"Item {item_id} deleted"}

@app.options("/items")
def options_items():
    """Handle OPTIONS requests (CORS preflight)."""
    return {"allowed_methods": ["GET", "POST", "PUT", "DELETE"]}

@app.head("/items")
def head_items():
    """Handle HEAD requests (like GET but no body)."""
    return None

1.3 Async vs Sync Handlers

FastAPI supports both synchronous and asynchronous handler functions. Use async def when your handler performs I/O-bound operations (database queries, HTTP calls, file operations):

import httpx
from fastapi import FastAPI

app = FastAPI()

# Synchronous handler — FastAPI runs this in a thread pool
@app.get("/sync")
def sync_handler():
    return {"type": "synchronous"}

# Asynchronous handler — runs directly on the event loop
@app.get("/async")
async def async_handler():
    return {"type": "asynchronous"}

# Real-world async example: calling an external API
@app.get("/external-data")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
        return response.json()

1.4 Route Order Matters

FastAPI evaluates routes in the order they are defined. This matters when you have routes with overlapping patterns:

from fastapi import FastAPI

app = FastAPI()

# This must come BEFORE "/users/{user_id}" to avoid "me" being
# interpreted as a user_id
@app.get("/users/me")
def read_current_user():
    return {"user": "current_user"}

@app.get("/users/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id}

If you reversed the order, a request to /users/me would match the /users/{user_id} route first and fail because "me" cannot be converted to an integer.

1.5 Multiple Decorators on One Function

You can apply multiple decorators to the same function if you want one handler to respond to multiple HTTP methods or paths:

from fastapi import FastAPI, APIRouter

app = FastAPI()

# Using api_route for multiple methods
@app.api_route("/health", methods=["GET", "HEAD"])
def health_check():
    return {"status": "healthy"}

1.6 Running the Application

# Development with auto-reload
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# The interactive docs are available at:
# http://localhost:8000/docs       (Swagger UI)
# http://localhost:8000/redoc      (ReDoc)

2. Path Parameters

Path parameters allow you to capture dynamic segments of the URL path. FastAPI uses Python type annotations to automatically parse, validate, and document path parameters.

2.1 Basic Path Parameters

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    """The user_id is automatically parsed as an integer."""
    return {"user_id": user_id}

@app.get("/files/{file_path:path}")
def get_file(file_path: str):
    """The :path converter captures the rest of the URL including slashes.
    Example: /files/home/user/document.txt
    """
    return {"file_path": file_path}

If a client sends GET /users/abc, FastAPI automatically returns a 422 validation error because "abc" cannot be converted to an integer:

# Automatic error response for invalid path parameter:
{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["path", "user_id"],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "abc"
        }
    ]
}

2.2 Type Annotations for Path Parameters

FastAPI supports various Python types for path parameters. The type annotation determines how the value is parsed and validated:

2.3 Enum Path Parameters

You can restrict path parameters to a set of predefined values using Python Enums:

from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class ModelName(str, Enum):
    """Available ML model names."""
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    messages = {
        ModelName.alexnet: "Deep Learning FTW!",
        ModelName.resnet: "LeCun would be proud",
        ModelName.lenet: "Have some residuals",
    }
    return {
        "model_name": model_name.value,
        "message": messages[model_name]
    }

2.4 Path() for Validation and Metadata

The Path() function provides additional validation and metadata for path parameters:

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(
        ...,
        title="Item ID",
        description="The unique identifier of the item to retrieve",
        ge=1,          # greater than or equal to 1
        le=10000,      # less than or equal to 10000
        examples=[42, 100, 999]
    )
):
    return {"item_id": item_id}

@app.get("/products/{product_code}")
def read_product(
    product_code: str = Path(
        ...,
        title="Product Code",
        description="Alphanumeric product code",
        min_length=3,
        max_length=10,
        pattern="^[A-Z]{2,4}-[0-9]{3,6}$"
    )
):
    """Accepts product codes like AB-123, PROD-456789."""
    return {"product_code": product_code}

2.5 Multiple Path Parameters

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/organizations/{org_id}/teams/{team_id}/members/{member_id}")
def get_team_member(
    org_id: int = Path(..., ge=1, description="Organization ID"),
    team_id: int = Path(..., ge=1, description="Team ID"),
    member_id: int = Path(..., ge=1, description="Member ID"),
):
    return {
        "org_id": org_id,
        "team_id": team_id,
        "member_id": member_id
    }

# UUID path parameters
from uuid import UUID

@app.get("/orders/{order_id}/items/{item_id}")
def get_order_item(order_id: UUID, item_id: UUID):
    return {
        "order_id": str(order_id),
        "item_id": str(item_id)
    }

3. Query Parameters

Query parameters are the key-value pairs that appear after the ? in a URL (e.g., /items?skip=0&limit=10). In FastAPI, any function parameter that is not a path parameter is automatically treated as a query parameter.

3.1 Basic Query Parameters

from fastapi import FastAPI

app = FastAPI()

# All three parameters are query parameters
# GET /items?skip=0&limit=10
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
    fake_items = [{"id": i, "name": f"Item {i}"} for i in range(100)]
    return fake_items[skip : skip + limit]

# Mix of path and query parameters
# GET /users/42/items?skip=0&limit=5
@app.get("/users/{user_id}/items")
def get_user_items(user_id: int, skip: int = 0, limit: int = 5):
    return {
        "user_id": user_id,
        "skip": skip,
        "limit": limit
    }

3.2 Required vs Optional Query Parameters

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/search")
def search(
    q: str,                              # Required (no default)
    category: str = "all",               # Optional with default "all"
    page: int = 1,                       # Optional with default 1
    include_archived: bool = False,      # Optional with default False
    sort_by: Optional[str] = None,       # Optional, defaults to None
):
    """
    GET /search?q=python                          (only required param)
    GET /search?q=python&category=books&page=2    (with optional params)
    GET /search?q=python&include_archived=true     (boolean query param)
    """
    results = {
        "query": q,
        "category": category,
        "page": page,
        "include_archived": include_archived,
    }
    if sort_by:
        results["sort_by"] = sort_by
    return results

3.3 Query() for Advanced Validation

The Query() function provides extensive validation options for query parameters:

from typing import Optional, List
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/products")
def search_products(
    # Required with validation
    q: str = Query(
        ...,
        min_length=2,
        max_length=100,
        title="Search Query",
        description="The search term to find products",
        examples=["laptop", "wireless mouse"]
    ),
    # Optional with numeric validation
    min_price: Optional[float] = Query(
        None,
        ge=0,
        description="Minimum price filter"
    ),
    max_price: Optional[float] = Query(
        None,
        ge=0,
        le=100000,
        description="Maximum price filter"
    ),
    # Pattern validation with regex
    sku: Optional[str] = Query(
        None,
        pattern="^[A-Z]{3}-[0-9]{4}$",
        description="Product SKU in format ABC-1234"
    ),
    # Pagination with defaults and limits
    page: int = Query(1, ge=1, le=1000, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page"),
):
    return {
        "query": q,
        "min_price": min_price,
        "max_price": max_price,
        "sku": sku,
        "page": page,
        "per_page": per_page,
    }

3.4 Multiple Values for a Query Parameter (Lists)

from typing import List, Optional
from fastapi import FastAPI, Query

app = FastAPI()

# GET /items?tag=python&tag=fastapi&tag=tutorial
@app.get("/items")
def filter_items(
    tag: List[str] = Query(
        default=[],
        title="Tags",
        description="Filter items by one or more tags",
    ),
    status: List[str] = Query(
        default=["active"],
        description="Filter by status. Defaults to active only.",
    ),
):
    return {
        "tags": tag,
        "statuses": status,
    }

# Enum-based list validation
from enum import Enum

class SortField(str, Enum):
    name = "name"
    price = "price"
    created_at = "created_at"
    rating = "rating"

@app.get("/catalog")
def get_catalog(
    sort_by: List[SortField] = Query(
        default=[SortField.created_at],
        description="Sort by one or more fields"
    ),
):
    return {"sort_by": [s.value for s in sort_by]}

3.5 Deprecating Query Parameters

from typing import Optional
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
def read_items(
    # Current parameter
    q: Optional[str] = Query(None, description="Search query"),
    # Deprecated parameter — still works but marked in docs
    search: Optional[str] = Query(
        None,
        deprecated=True,
        description="Deprecated. Use 'q' instead."
    ),
):
    query = q or search
    return {"query": query}

3.6 Alias for Query Parameters

from fastapi import FastAPI, Query

app = FastAPI()

# When the query param name isn't a valid Python identifier
# GET /items?item-query=something
@app.get("/items")
def read_items(
    q: str = Query(
        ...,
        alias="item-query",
        description="The query string uses a hyphen in the URL"
    )
):
    return {"query": q}

4. Request Headers

HTTP headers carry metadata about the request. FastAPI provides the Header() function to extract and validate header values with full type support.

4.1 Basic Header Extraction

from typing import Optional
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items")
def read_items(
    user_agent: Optional[str] = Header(None),
    accept_language: Optional[str] = Header(None),
    x_request_id: Optional[str] = Header(None),
):
    """FastAPI automatically converts header names:
    - user_agent matches the 'User-Agent' header
    - accept_language matches the 'Accept-Language' header
    - x_request_id matches the 'X-Request-Id' header

    Underscores in the Python parameter become hyphens in the HTTP header.
    """
    return {
        "user_agent": user_agent,
        "accept_language": accept_language,
        "x_request_id": x_request_id,
    }

4.2 Required Headers and Validation

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/secure-data")
def get_secure_data(
    x_api_key: str = Header(
        ...,
        description="API key for authentication",
        min_length=20,
        max_length=100,
    ),
    x_api_version: str = Header(
        "v1",
        description="API version",
        pattern="^v[0-9]+$"
    ),
):
    # Validate API key
    if x_api_key != "my-secret-key-that-is-long-enough":
        raise HTTPException(status_code=403, detail="Invalid API key")
    return {"data": "sensitive information", "api_version": x_api_version}

4.3 Duplicate Headers (Multiple Values)

from typing import List, Optional
from fastapi import FastAPI, Header

app = FastAPI()

# Some headers can appear multiple times (e.g., X-Forwarded-For)
@app.get("/check-proxies")
def check_proxies(
    x_forwarded_for: Optional[List[str]] = Header(None)
):
    return {"proxies": x_forwarded_for}

4.4 Common Header Patterns

from typing import Optional
from fastapi import FastAPI, Header, HTTPException, Depends

app = FastAPI()

# Pattern: Extract and validate Authorization header
def get_token_from_header(
    authorization: str = Header(..., description="Bearer token")
) -> str:
    """Extract token from Authorization header."""
    if not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=401,
            detail="Authorization header must start with 'Bearer '"
        )
    return authorization[7:]  # Strip "Bearer " prefix

@app.get("/profile")
def get_profile(token: str = Depends(get_token_from_header)):
    return {"token": token, "user": "authenticated_user"}

# Pattern: Content negotiation
@app.get("/data")
def get_data(accept: Optional[str] = Header("application/json")):
    if "text/html" in accept:
        return {"format": "Would return HTML"}
    elif "application/xml" in accept:
        return {"format": "Would return XML"}
    return {"format": "JSON", "data": [1, 2, 3]}

# Pattern: Rate limiting headers
@app.get("/rate-limited")
def rate_limited_endpoint(
    x_client_id: str = Header(..., description="Client identifier for rate limiting"),
):
    return {
        "client_id": x_client_id,
        "message": "Request processed"
    }

5. Request Body

For POST, PUT, and PATCH requests, clients typically send data in the request body. FastAPI uses Pydantic models to define, validate, and document request body schemas.

5.1 Basic Request Body with Pydantic

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class ItemCreate(BaseModel):
    """Schema for creating a new item."""
    name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="The name of the item",
        examples=["Widget Pro"]
    )
    description: Optional[str] = Field(
        None,
        max_length=1000,
        description="Optional item description"
    )
    price: float = Field(
        ...,
        gt=0,
        le=1000000,
        description="Price in USD"
    )
    tax: Optional[float] = Field(
        None,
        ge=0,
        le=100,
        description="Tax percentage"
    )
    tags: list[str] = Field(
        default_factory=list,
        max_length=10,
        description="List of tags"
    )

@app.post("/items")
def create_item(item: ItemCreate):
    """
    Send a JSON body:
    {
        "name": "Widget Pro",
        "price": 29.99,
        "tax": 8.5,
        "tags": ["electronics", "gadget"]
    }
    """
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + (item.price * item.tax / 100)
        item_dict["price_with_tax"] = round(price_with_tax, 2)
    return item_dict

5.2 Nested Models

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

app = FastAPI()

class Address(BaseModel):
    street: str = Field(..., min_length=1)
    city: str = Field(..., min_length=1)
    state: str = Field(..., min_length=2, max_length=2)
    zip_code: str = Field(..., pattern="^[0-9]{5}(-[0-9]{4})?$")
    country: str = Field(default="US")

class ContactInfo(BaseModel):
    email: str = Field(..., description="Contact email")
    phone: Optional[str] = Field(None, pattern="^\+?[0-9\-\s()]+$")

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

class OrderCreate(BaseModel):
    customer_name: str = Field(..., min_length=1, max_length=200)
    contact: ContactInfo
    shipping_address: Address
    billing_address: Optional[Address] = None
    items: List[OrderItem] = Field(..., min_length=1)
    notes: Optional[str] = None

    class Config:
        json_schema_extra = {
            "example": {
                "customer_name": "Jane Doe",
                "contact": {
                    "email": "jane@example.com",
                    "phone": "+1-555-0100"
                },
                "shipping_address": {
                    "street": "123 Main St",
                    "city": "Springfield",
                    "state": "IL",
                    "zip_code": "62701"
                },
                "items": [
                    {"product_id": 1, "quantity": 2, "unit_price": 29.99}
                ],
                "notes": "Leave at door"
            }
        }

@app.post("/orders")
def create_order(order: OrderCreate):
    total = sum(item.quantity * item.unit_price for item in order.items)
    billing = order.billing_address or order.shipping_address
    return {
        "order_id": 12345,
        "customer": order.customer_name,
        "total": round(total, 2),
        "item_count": len(order.items),
        "billing_city": billing.city,
        "created_at": datetime.utcnow().isoformat()
    }

5.3 Multiple Body Parameters

from fastapi import FastAPI, Body
from pydantic import BaseModel, Field

app = FastAPI()

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

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

@app.put("/items/{item_id}")
def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(
        ...,
        ge=1,
        le=5,
        description="Priority level of the update"
    ),
    note: str = Body(
        None,
        max_length=500,
        description="Optional note about the update"
    ),
):
    """
    Expected JSON body:
    {
        "item": {"name": "Widget", "price": 9.99},
        "user": {"username": "john", "email": "john@example.com"},
        "importance": 3,
        "note": "Updated pricing"
    }
    """
    return {
        "item_id": item_id,
        "item": item.model_dump(),
        "user": user.model_dump(),
        "importance": importance,
        "note": note
    }

5.4 Embedding a Single Body Parameter

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

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

# Without embed=True: {"name": "Widget", "price": 9.99}
# With embed=True:    {"item": {"name": "Widget", "price": 9.99}}
@app.post("/items")
def create_item(item: Item = Body(..., embed=True)):
    return item

5.5 Raw Body Access

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def receive_webhook(request: Request):
    """Access raw request body for webhook signature verification."""
    raw_body = await request.body()
    json_body = await request.json()

    # Useful for webhook signature verification
    content_type = request.headers.get("content-type", "")

    return {
        "content_type": content_type,
        "body_size": len(raw_body),
        "parsed": json_body
    }

6. Cookie Parameters

FastAPI provides the Cookie() function to extract cookies from incoming requests, with the same validation and documentation features as other parameter types.

from typing import Optional
from fastapi import FastAPI, Cookie

app = FastAPI()

@app.get("/items")
def read_items(
    session_id: Optional[str] = Cookie(None, description="Session cookie"),
    tracking_id: Optional[str] = Cookie(None, description="Analytics tracking ID"),
    preferences: Optional[str] = Cookie(None, description="User preferences JSON"),
):
    return {
        "session_id": session_id,
        "tracking_id": tracking_id,
        "preferences": preferences,
    }

# Real-world example: session-based user identification
@app.get("/dashboard")
def get_dashboard(
    session_token: str = Cookie(
        ...,
        min_length=32,
        max_length=128,
        description="Required session token"
    ),
):
    # In production, you would validate this against a session store
    return {"session": session_token, "message": "Welcome to dashboard"}

curl -b "session_id=abc123;tracking_id=xyz789" http://localhost:8000/items

7. Form Data and File Uploads

For HTML forms and file uploads, FastAPI provides the Form() and File()/UploadFile types. These require the python-multipart package.

7.1 Form Data

from fastapi import FastAPI, Form

app = FastAPI()

# Standard HTML form submission
@app.post("/login")
def login(
    username: str = Form(
        ...,
        min_length=3,
        max_length=50,
        description="Username"
    ),
    password: str = Form(
        ...,
        min_length=8,
        description="Password"
    ),
    remember_me: bool = Form(False),
):
    """
    Handles application/x-www-form-urlencoded data.
    Test with:
    curl -X POST http://localhost:8000/login \
         -d "username=admin&password=secret123&remember_me=true"
    """
    return {
        "username": username,
        "remember_me": remember_me,
        "message": "Login successful"
    }

# OAuth2 token endpoint (real-world pattern)
@app.post("/token")
def get_token(
    grant_type: str = Form(..., pattern="^(password|refresh_token|client_credentials)$"),
    username: str = Form(None),
    password: str = Form(None),
    refresh_token: str = Form(None),
    scope: str = Form(""),
):
    if grant_type == "password":
        if not username or not password:
            return {"error": "username and password required for password grant"}
    return {
        "access_token": "fake-jwt-token",
        "token_type": "bearer",
        "scope": scope
    }

7.2 File Uploads

from typing import List
from fastapi import FastAPI, File, UploadFile, HTTPException, Form

app = FastAPI()

# Simple file upload
@app.post("/upload")
async def upload_file(file: UploadFile = File(..., description="File to upload")):
    """
    UploadFile provides:
    - file.filename: Original filename
    - file.content_type: MIME type (e.g., "image/png")
    - file.size: File size in bytes
    - file.read(): Read file contents (async)
    - file.seek(0): Reset file position
    - file.close(): Close the file
    """
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents),
    }

# File upload with validation
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_SIZE = 5 * 1024 * 1024  # 5MB

@app.post("/upload-image")
async def upload_image(
    image: UploadFile = File(..., description="Image file (JPEG, PNG, GIF, WebP)"),
    alt_text: str = Form("", max_length=200),
):
    # Validate content type
    if image.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid file type: {image.content_type}. Allowed: {ALLOWED_TYPES}"
        )

    # Read and validate size
    contents = await image.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(
            status_code=400,
            detail=f"File too large: {len(contents)} bytes. Max: {MAX_SIZE} bytes"
        )

    # In production, save to disk or cloud storage
    return {
        "filename": image.filename,
        "content_type": image.content_type,
        "size": len(contents),
        "alt_text": alt_text,
    }

# Multiple file upload
@app.post("/upload-multiple")
async def upload_multiple_files(
    files: List[UploadFile] = File(..., description="Multiple files to upload"),
):
    results = []
    for f in files:
        contents = await f.read()
        results.append({
            "filename": f.filename,
            "content_type": f.content_type,
            "size": len(contents),
        })
    return {"files_uploaded": len(results), "files": results}

# Single file
curl -X POST http://localhost:8000/upload -F "file=@photo.jpg"

# Multiple files
curl -X POST http://localhost:8000/upload-multiple      -F "files=@photo1.jpg" -F "files=@photo2.png"

# File with form data
curl -X POST http://localhost:8000/upload-image      -F "image=@photo.jpg" -F "alt_text=My vacation photo"

8. Response Handling

While FastAPI defaults to returning JSON responses, it supports multiple response types for different use cases.

8.1 JSONResponse (Default)

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# Default behavior — dict is automatically serialized to JSON
@app.get("/items")
def get_items():
    return {"items": [{"id": 1, "name": "Widget"}]}

# Explicit JSONResponse with custom status code and headers
@app.post("/items")
def create_item():
    return JSONResponse(
        content={"id": 1, "name": "Widget", "message": "Created"},
        status_code=201,
        headers={"X-Custom-Header": "created"}
    )

# Custom JSON encoder for special types
import json
from datetime import datetime, date
from decimal import Decimal

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        return super().default(obj)

@app.get("/financial-report")
def financial_report():
    data = {
        "total": Decimal("1234567.89"),
        "generated_at": datetime.utcnow(),
    }
    return JSONResponse(
        content=json.loads(json.dumps(data, cls=CustomEncoder))
    )

8.2 HTMLResponse

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/page", response_class=HTMLResponse)
def get_page():
    return """
    <html>
        <head><title>FastAPI HTML</title></head>
        <body>
            <h1>Hello from FastAPI!</h1>
            <p>This is an HTML response.</p>
        </body>
    </html>
    """

# Dynamic HTML with data
@app.get("/users/{user_id}/profile", response_class=HTMLResponse)
def user_profile(user_id: int):
    return f"""
    <html>
        <body>
            <h1>User Profile</h1>
            <p>User ID: {user_id}</p>
        </body>
    </html>
    """

8.3 RedirectResponse

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/old-page")
def old_page():
    """301 permanent redirect."""
    return RedirectResponse(url="/new-page", status_code=301)

@app.get("/new-page")
def new_page():
    return {"message": "This is the new page"}

@app.post("/submit-form")
def submit_form():
    """303 See Other — redirect after POST (PRG pattern)."""
    # Process form data...
    return RedirectResponse(url="/success", status_code=303)

@app.get("/external")
def external_redirect():
    """Redirect to an external URL."""
    return RedirectResponse(url="https://fastapi.tiangolo.com")

8.4 StreamingResponse

import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

# Stream a large CSV file
def generate_csv():
    """Generator that yields CSV rows one at a time."""
    yield "id,name,email\n"
    for i in range(1, 10001):
        yield f"{i},user_{i},user_{i}@example.com\n"

@app.get("/export/csv")
def export_csv():
    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

# Server-Sent Events (SSE) for real-time updates
async def event_stream() -> AsyncGenerator[str, None]:
    """Generates Server-Sent Events."""
    for i in range(10):
        await asyncio.sleep(1)
        yield f"data: {{"count": {i}, "message": "Update {i}"}}\n\n"
    yield "data: {"message": "Stream complete"}\n\n"

@app.get("/events")
async def get_events():
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

8.5 FileResponse

import os
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse

app = FastAPI()

UPLOAD_DIR = "/tmp/uploads"

@app.get("/download/{filename}")
def download_file(filename: str):
    file_path = os.path.join(UPLOAD_DIR, filename)

    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream",
    )

# Inline display (e.g., images)
@app.get("/images/{image_name}")
def get_image(image_name: str):
    file_path = os.path.join(UPLOAD_DIR, "images", image_name)

    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="Image not found")

    return FileResponse(
        path=file_path,
        media_type="image/png",
        # No Content-Disposition header = inline display
    )

8.6 PlainTextResponse and Custom Responses

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse, Response

app = FastAPI()

@app.get("/robots.txt", response_class=PlainTextResponse)
def robots_txt():
    return """User-agent: *
Disallow: /admin/
Allow: /
Sitemap: https://example.com/sitemap.xml"""

# XML response
@app.get("/sitemap.xml")
def sitemap():
    xml_content = """<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
    <url><loc>https://example.com/</loc></url>
</urlset>"""
    return Response(
        content=xml_content,
        media_type="application/xml"
    )

9. Status Codes

Proper HTTP status codes are essential for RESTful APIs. FastAPI provides the status module with named constants for all standard HTTP status codes.

9.1 Setting Status Codes

from fastapi import FastAPI, status

app = FastAPI()

# Set status code in the decorator
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item():
    return {"id": 1, "name": "Widget"}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
    # 204 responses should not have a body
    return None

@app.get("/items", status_code=status.HTTP_200_OK)
def list_items():
    return {"items": []}

9.2 Common Status Code Constants

9.3 Dynamic Status Codes with Response Parameter

from fastapi import FastAPI, Response, status

app = FastAPI()

items_db = {}

@app.put("/items/{item_id}")
def upsert_item(item_id: int, response: Response):
    """Create or update an item. Returns 201 for creation, 200 for update."""
    if item_id in items_db:
        items_db[item_id] = {"id": item_id, "updated": True}
        response.status_code = status.HTTP_200_OK
        return {"message": "Item updated", "item": items_db[item_id]}
    else:
        items_db[item_id] = {"id": item_id, "updated": False}
        response.status_code = status.HTTP_201_CREATED
        return {"message": "Item created", "item": items_db[item_id]}

9.4 HTTPException for Error Responses

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items_db = {
    1: {"id": 1, "name": "Widget", "owner": "alice"},
    2: {"id": 2, "name": "Gadget", "owner": "bob"},
}

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with ID {item_id} not found",
            headers={"X-Error-Code": "ITEM_NOT_FOUND"},
        )
    return items_db[item_id]

@app.delete("/items/{item_id}")
def delete_item(item_id: int, current_user: str = "alice"):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="Item not found"
        )
    item = items_db[item_id]
    if item["owner"] != current_user:
        raise HTTPException(
            status_code=403,
            detail="You do not have permission to delete this item"
        )
    del items_db[item_id]
    return {"message": f"Item {item_id} deleted"}

10. Response Headers and Cookies

10.1 Setting Response Headers

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse

app = FastAPI()

# Method 1: Using the Response parameter
@app.get("/items")
def get_items(response: Response):
    response.headers["X-Total-Count"] = "42"
    response.headers["X-Page"] = "1"
    response.headers["X-Per-Page"] = "10"
    response.headers["Cache-Control"] = "public, max-age=300"
    return {"items": [{"id": 1}]}

# Method 2: Using JSONResponse directly
@app.get("/data")
def get_data():
    return JSONResponse(
        content={"data": "value"},
        headers={
            "X-Request-Id": "abc-123",
            "X-Rate-Limit-Remaining": "99",
            "X-Rate-Limit-Reset": "1609459200",
        }
    )

# Method 3: CORS and security headers via middleware
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware for security headers
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = "max-age=31536000"
        return response

app.add_middleware(SecurityHeadersMiddleware)

10.2 Cookie Management

from datetime import datetime, timedelta
from fastapi import FastAPI, Response, Cookie
from typing import Optional

app = FastAPI()

# Setting cookies
@app.post("/login")
def login(response: Response):
    # Secure session cookie
    response.set_cookie(
        key="session_id",
        value="abc123def456",
        httponly=True,       # Not accessible via JavaScript
        secure=True,         # Only sent over HTTPS
        samesite="lax",      # CSRF protection
        max_age=3600,        # Expires in 1 hour
        path="/",            # Available on all paths
        domain=None,         # Current domain only
    )

    # Non-sensitive preference cookie
    response.set_cookie(
        key="theme",
        value="dark",
        httponly=False,      # Accessible via JavaScript
        max_age=86400 * 365, # Expires in 1 year
    )

    return {"message": "Login successful"}

# Reading cookies
@app.get("/dashboard")
def dashboard(
    session_id: Optional[str] = Cookie(None),
    theme: Optional[str] = Cookie("light"),
):
    if not session_id:
        return {"error": "Not authenticated"}
    return {
        "session": session_id,
        "theme": theme,
        "message": "Welcome to dashboard"
    }

# Deleting cookies
@app.post("/logout")
def logout(response: Response):
    response.delete_cookie(key="session_id", path="/")
    response.delete_cookie(key="theme")
    return {"message": "Logged out"}

11. APIRouter — Organizing Routes

As your application grows, keeping all routes in a single file becomes unmanageable. FastAPI’s APIRouter lets you organize routes into separate modules with their own prefixes, tags, and dependencies — similar to Flask’s Blueprints.

11.1 Project Structure

myproject/
├── main.py              # Application entry point
├── routers/
│   ├── __init__.py
│   ├── users.py         # User-related routes
│   ├── items.py         # Item-related routes
│   └── auth.py          # Authentication routes
├── models/
│   ├── __init__.py
│   ├── user.py          # Pydantic models for users
│   └── item.py          # Pydantic models for items
└── dependencies.py      # Shared dependencies

11.2 Creating Routers

# routers/users.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List

router = APIRouter(
    prefix="/users",
    tags=["Users"],
    responses={
        404: {"description": "User not found"},
    },
)

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(...)
    full_name: Optional[str] = None

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    full_name: Optional[str] = None

# In-memory database for demonstration
users_db: dict[int, dict] = {}
next_id = 1

@router.get("/", response_model=List[UserResponse])
def list_users(skip: int = 0, limit: int = 10):
    """List all users with pagination."""
    users = list(users_db.values())
    return users[skip : skip + limit]

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    """Create a new user."""
    global next_id
    user_data = {"id": next_id, **user.model_dump()}
    users_db[next_id] = user_data
    next_id += 1
    return user_data

@router.get("/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    """Get a user by ID."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

@router.put("/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserCreate):
    """Update a user."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    users_db[user_id] = {"id": user_id, **user.model_dump()}
    return users_db[user_id]

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int):
    """Delete a user."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]

# routers/items.py
from fastapi import APIRouter, HTTPException, Depends, status
from pydantic import BaseModel, Field
from typing import Optional, List

router = APIRouter(
    prefix="/items",
    tags=["Items"],
)

class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = None
    price: float = Field(..., gt=0)
    category: str = Field(..., min_length=1)

class ItemResponse(BaseModel):
    id: int
    name: str
    description: Optional[str]
    price: float
    category: str
    owner_id: int

items_db: dict[int, dict] = {}
next_id = 1

@router.get("/", response_model=List[ItemResponse])
def list_items(
    category: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    skip: int = 0,
    limit: int = 20,
):
    """List items with optional filters."""
    items = list(items_db.values())

    if category:
        items = [i for i in items if i["category"] == category]
    if min_price is not None:
        items = [i for i in items if i["price"] >= min_price]
    if max_price is not None:
        items = [i for i in items if i["price"] <= max_price]

    return items[skip : skip + limit]

@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate):
    """Create a new item."""
    global next_id
    item_data = {"id": next_id, "owner_id": 1, **item.model_dump()}
    items_db[next_id] = item_data
    next_id += 1
    return item_data

@router.get("/{item_id}", response_model=ItemResponse)
def get_item(item_id: int):
    """Get an item by ID."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

11.3 Including Routers in the Main Application

# main.py
from fastapi import FastAPI
from routers import users, items

app = FastAPI(
    title="My Application",
    description="A well-organized FastAPI application",
    version="1.0.0",
)

# Include routers
app.include_router(users.router)
app.include_router(items.router)

# You can also override prefix and tags when including
# app.include_router(
#     users.router,
#     prefix="/api/v1/users",   # Override prefix
#     tags=["Users V1"],        # Override tags
# )

@app.get("/")
def root():
    return {"message": "Welcome to the API"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

11.4 Nested Routers

# routers/admin/__init__.py
from fastapi import APIRouter
from . import dashboard, settings

router = APIRouter(prefix="/admin", tags=["Admin"])
router.include_router(dashboard.router)
router.include_router(settings.router)

# routers/admin/dashboard.py
from fastapi import APIRouter

router = APIRouter(prefix="/dashboard")

@router.get("/")
def admin_dashboard():
    """GET /admin/dashboard/"""
    return {"message": "Admin dashboard"}

@router.get("/stats")
def admin_stats():
    """GET /admin/dashboard/stats"""
    return {"users": 100, "items": 500}

# routers/admin/settings.py
from fastapi import APIRouter

router = APIRouter(prefix="/settings")

@router.get("/")
def admin_settings():
    """GET /admin/settings/"""
    return {"theme": "dark", "language": "en"}

11.5 Router with Shared Dependencies

from fastapi import APIRouter, Depends, Header, HTTPException

# Dependency: require API key for all routes in this router
async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != "secret-api-key":
        raise HTTPException(status_code=403, detail="Invalid API key")
    return x_api_key

# All routes in this router require the API key
router = APIRouter(
    prefix="/internal",
    tags=["Internal"],
    dependencies=[Depends(verify_api_key)],
)

@router.get("/metrics")
def get_metrics():
    return {"cpu": 45.2, "memory": 68.1}

@router.get("/logs")
def get_logs():
    return {"logs": ["entry1", "entry2"]}

12. Route Dependencies

FastAPI’s dependency injection system is one of its most powerful features. Dependencies let you share logic across routes — authentication, database connections, pagination, logging, and more.

12.1 Basic Dependencies with Depends()

from fastapi import FastAPI, Depends, Query

app = FastAPI()

# Dependency function for common pagination parameters
def pagination_params(
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page"),
):
    """Reusable pagination dependency."""
    skip = (page - 1) * per_page
    return {"skip": skip, "limit": per_page, "page": page}

@app.get("/users")
def list_users(pagination: dict = Depends(pagination_params)):
    fake_users = [{"id": i, "name": f"User {i}"} for i in range(100)]
    users = fake_users[pagination["skip"]: pagination["skip"] + pagination["limit"]]
    return {
        "page": pagination["page"],
        "per_page": pagination["limit"],
        "users": users
    }

@app.get("/items")
def list_items(pagination: dict = Depends(pagination_params)):
    fake_items = [{"id": i, "name": f"Item {i}"} for i in range(200)]
    items = fake_items[pagination["skip"]: pagination["skip"] + pagination["limit"]]
    return {
        "page": pagination["page"],
        "per_page": pagination["limit"],
        "items": items
    }

12.2 Class-Based Dependencies

from fastapi import FastAPI, Depends, Query
from typing import Optional

app = FastAPI()

class PaginationParams:
    """Callable class that works as a dependency."""

    def __init__(
        self,
        page: int = Query(1, ge=1),
        per_page: int = Query(20, ge=1, le=100),
        sort_by: Optional[str] = Query(None),
        order: str = Query("asc", pattern="^(asc|desc)$"),
    ):
        self.page = page
        self.per_page = per_page
        self.skip = (page - 1) * per_page
        self.sort_by = sort_by
        self.order = order

@app.get("/products")
def list_products(params: PaginationParams = Depends()):
    """When using Depends() without arguments on a class,
    FastAPI uses the class itself as the dependency."""
    return {
        "page": params.page,
        "per_page": params.per_page,
        "sort_by": params.sort_by,
        "order": params.order,
    }

12.3 Dependency Chains

from fastapi import FastAPI, Depends, HTTPException, Header, status

app = FastAPI()

# Level 1: Extract token
def get_token(authorization: str = Header(...)) -> str:
    if not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authorization header"
        )
    return authorization[7:]

# Level 2: Decode token and get user (depends on Level 1)
def get_current_user(token: str = Depends(get_token)) -> dict:
    # In production, decode JWT and query database
    if token == "valid-token":
        return {"id": 1, "username": "alice", "role": "admin"}
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or expired token"
    )

# Level 3: Check admin role (depends on Level 2)
def require_admin(user: dict = Depends(get_current_user)) -> dict:
    if user["role"] != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return user

# Public route — no authentication needed
@app.get("/public")
def public_endpoint():
    return {"message": "Anyone can access this"}

# Authenticated route — requires valid token
@app.get("/profile")
def get_profile(user: dict = Depends(get_current_user)):
    return {"user": user}

# Admin-only route — requires admin role
@app.get("/admin/users")
def admin_list_users(admin: dict = Depends(require_admin)):
    return {"admin": admin["username"], "users": ["alice", "bob", "charlie"]}

12.4 Dependencies with Yield (Context Managers)

from fastapi import FastAPI, Depends
from typing import Generator

app = FastAPI()

# Simulating a database session
class DatabaseSession:
    def __init__(self):
        self.connected = True
        print("Database session opened")

    def query(self, sql: str):
        return [{"id": 1}]

    def close(self):
        self.connected = False
        print("Database session closed")

def get_db() -> Generator[DatabaseSession, None, None]:
    """Dependency with cleanup — session is closed after the request."""
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
def get_users(db: DatabaseSession = Depends(get_db)):
    """The database session is automatically closed after this handler completes,
    even if an exception occurs."""
    results = db.query("SELECT * FROM users")
    return {"users": results}

12.5 Application-Level Dependencies

from fastapi import FastAPI, Depends, Header, HTTPException
import time

# Dependency applied to ALL routes
async def log_request(x_request_id: str = Header(None)):
    """Log every incoming request."""
    request_id = x_request_id or f"auto-{int(time.time())}"
    print(f"Request ID: {request_id}")
    return request_id

app = FastAPI(dependencies=[Depends(log_request)])

@app.get("/")
def root():
    return {"message": "This route also has the log_request dependency"}

@app.get("/items")
def items():
    return {"message": "So does this one"}

13. The Request Object

Sometimes you need access to the raw request object for information that is not covered by FastAPI’s parameter declarations. You can access it by adding a parameter of type Request.

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/request-info")
async def request_info(request: Request):
    """Access detailed information about the incoming request."""
    return {
        # URL information
        "url": str(request.url),
        "base_url": str(request.base_url),
        "path": request.url.path,
        "query_string": str(request.query_params),

        # Client information
        "client_host": request.client.host if request.client else None,
        "client_port": request.client.port if request.client else None,

        # Request metadata
        "method": request.method,
        "headers": dict(request.headers),
        "cookies": request.cookies,

        # Path parameters
        "path_params": request.path_params,
    }

@app.post("/webhook")
async def webhook(request: Request):
    """Read raw body for webhook signature verification."""
    body = await request.body()
    content_type = request.headers.get("content-type", "")

    if "application/json" in content_type:
        json_body = await request.json()
        return {"type": "json", "data": json_body}
    else:
        return {"type": "raw", "size": len(body)}

# Practical: Request-scoped state
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
import time

class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Attach metadata to request.state for use in route handlers
        request.state.request_id = str(uuid.uuid4())
        request.state.start_time = time.time()
        response = await call_next(request)
        duration = time.time() - request.state.start_time
        response.headers["X-Request-Id"] = request.state.request_id
        response.headers["X-Process-Time"] = f"{duration:.4f}"
        return response

app.add_middleware(RequestContextMiddleware)

@app.get("/with-context")
async def with_context(request: Request):
    return {
        "request_id": request.state.request_id,
        "message": "This request has context from middleware"
    }

14. Practical Example — URL Shortener Service

Let us bring everything together by building a complete URL shortener service. This practical example demonstrates path parameters, query parameters, request bodies, headers, cookies, multiple response types, dependency injection, and API routers — all in one application.

14.1 Project Structure

url_shortener/
├── main.py              # Application entry point and configuration
├── models.py            # Pydantic models
├── database.py          # In-memory database (replace with real DB in production)
├── dependencies.py      # Shared dependencies
├── routers/
│   ├── __init__.py
│   ├── urls.py          # URL shortening routes
│   ├── analytics.py     # Analytics routes
│   └── admin.py         # Admin routes
└── requirements.txt

14.2 Models

# models.py
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, HttpUrl

class URLCreate(BaseModel):
    """Request body for creating a shortened URL."""
    original_url: str = Field(
        ...,
        min_length=10,
        max_length=2048,
        description="The original URL to shorten",
        examples=["https://www.example.com/very/long/path"]
    )
    custom_code: Optional[str] = Field(
        None,
        min_length=3,
        max_length=20,
        pattern="^[a-zA-Z0-9_-]+$",
        description="Custom short code (optional)"
    )
    expires_in_hours: Optional[int] = Field(
        None,
        ge=1,
        le=8760,  # Max 1 year
        description="Hours until the link expires"
    )
    tags: List[str] = Field(
        default_factory=list,
        max_length=5,
        description="Tags for organizing URLs"
    )

class URLResponse(BaseModel):
    """Response for a shortened URL."""
    short_code: str
    short_url: str
    original_url: str
    created_at: str
    expires_at: Optional[str] = None
    click_count: int = 0
    tags: List[str] = []

class URLStats(BaseModel):
    """Analytics data for a shortened URL."""
    short_code: str
    original_url: str
    total_clicks: int
    unique_visitors: int
    clicks_by_day: dict
    top_referrers: List[dict]
    top_countries: List[dict]
    created_at: str

class URLBulkCreate(BaseModel):
    """Request body for bulk URL creation."""
    urls: List[URLCreate] = Field(
        ...,
        min_length=1,
        max_length=100,
        description="List of URLs to shorten"
    )

14.3 Database Module

# database.py
import string
import random
from datetime import datetime, timedelta
from typing import Optional, Dict, List
from collections import defaultdict

class URLDatabase:
    """In-memory URL database. Replace with a real database in production."""

    def __init__(self):
        self.urls: Dict[str, dict] = {}
        self.clicks: Dict[str, List[dict]] = defaultdict(list)

    def generate_code(self, length: int = 6) -> str:
        """Generate a random short code."""
        chars = string.ascii_letters + string.digits
        while True:
            code = "".join(random.choices(chars, k=length))
            if code not in self.urls:
                return code

    def create_url(
        self,
        original_url: str,
        custom_code: Optional[str] = None,
        expires_in_hours: Optional[int] = None,
        tags: Optional[List[str]] = None,
    ) -> dict:
        """Create a new shortened URL."""
        code = custom_code or self.generate_code()

        if code in self.urls:
            raise ValueError(f"Code '{code}' is already taken")

        now = datetime.utcnow()
        expires_at = None
        if expires_in_hours:
            expires_at = now + timedelta(hours=expires_in_hours)

        url_data = {
            "short_code": code,
            "original_url": original_url,
            "created_at": now.isoformat(),
            "expires_at": expires_at.isoformat() if expires_at else None,
            "click_count": 0,
            "tags": tags or [],
            "is_active": True,
        }
        self.urls[code] = url_data
        return url_data

    def get_url(self, code: str) -> Optional[dict]:
        """Get URL data by short code."""
        url_data = self.urls.get(code)
        if not url_data:
            return None

        # Check expiration
        if url_data["expires_at"]:
            expires_at = datetime.fromisoformat(url_data["expires_at"])
            if datetime.utcnow() > expires_at:
                url_data["is_active"] = False
                return None

        return url_data

    def record_click(self, code: str, visitor_ip: str, referrer: str, user_agent: str):
        """Record a click event."""
        if code in self.urls:
            self.urls[code]["click_count"] += 1
            self.clicks[code].append({
                "timestamp": datetime.utcnow().isoformat(),
                "visitor_ip": visitor_ip,
                "referrer": referrer,
                "user_agent": user_agent,
            })

    def get_stats(self, code: str) -> Optional[dict]:
        """Get analytics for a URL."""
        url_data = self.urls.get(code)
        if not url_data:
            return None

        clicks = self.clicks.get(code, [])
        unique_ips = set(c["visitor_ip"] for c in clicks)

        # Group clicks by day
        clicks_by_day = defaultdict(int)
        for click in clicks:
            day = click["timestamp"][:10]
            clicks_by_day[day] += 1

        # Top referrers
        referrer_counts = defaultdict(int)
        for click in clicks:
            ref = click["referrer"] or "Direct"
            referrer_counts[ref] += 1
        top_referrers = [
            {"referrer": r, "count": c}
            for r, c in sorted(referrer_counts.items(), key=lambda x: -x[1])[:10]
        ]

        return {
            "short_code": code,
            "original_url": url_data["original_url"],
            "total_clicks": url_data["click_count"],
            "unique_visitors": len(unique_ips),
            "clicks_by_day": dict(clicks_by_day),
            "top_referrers": top_referrers,
            "top_countries": [],  # Would require GeoIP in production
            "created_at": url_data["created_at"],
        }

    def list_urls(
        self,
        skip: int = 0,
        limit: int = 20,
        tag: Optional[str] = None,
    ) -> List[dict]:
        """List all URLs with optional filtering."""
        urls = list(self.urls.values())
        if tag:
            urls = [u for u in urls if tag in u["tags"]]
        return urls[skip : skip + limit]

    def delete_url(self, code: str) -> bool:
        """Delete a URL."""
        if code in self.urls:
            del self.urls[code]
            self.clicks.pop(code, None)
            return True
        return False


# Singleton database instance
db = URLDatabase()

14.4 Dependencies

# dependencies.py
from fastapi import Depends, Header, HTTPException, Query, Request, status
from database import URLDatabase, db
from typing import Optional

def get_database() -> URLDatabase:
    """Provide the database instance."""
    return db

class PaginationParams:
    """Reusable pagination dependency."""

    def __init__(
        self,
        page: int = Query(1, ge=1, description="Page number"),
        per_page: int = Query(20, ge=1, le=100, description="Items per page"),
    ):
        self.page = page
        self.per_page = per_page
        self.skip = (page - 1) * per_page

def get_api_key(x_api_key: str = Header(..., description="API key")) -> str:
    """Validate API key for protected routes."""
    valid_keys = {"admin-key-001", "admin-key-002"}
    if x_api_key not in valid_keys:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API key"
        )
    return x_api_key

def get_client_info(request: Request) -> dict:
    """Extract client information from the request."""
    return {
        "ip": request.client.host if request.client else "unknown",
        "user_agent": request.headers.get("user-agent", "unknown"),
        "referrer": request.headers.get("referer", ""),
    }

14.5 URL Routes

# routers/urls.py
from fastapi import APIRouter, Depends, HTTPException, Response, status, Cookie
from fastapi.responses import RedirectResponse, JSONResponse
from typing import Optional, List
from models import URLCreate, URLResponse, URLBulkCreate
from database import URLDatabase
from dependencies import get_database, PaginationParams, get_client_info

router = APIRouter(tags=["URLs"])

BASE_URL = "http://localhost:8000"

@router.post(
    "/shorten",
    response_model=URLResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a shortened URL",
)
def create_short_url(
    url_data: URLCreate,
    response: Response,
    db: URLDatabase = Depends(get_database),
):
    """Create a new shortened URL with optional custom code and expiration."""
    try:
        url = db.create_url(
            original_url=url_data.original_url,
            custom_code=url_data.custom_code,
            expires_in_hours=url_data.expires_in_hours,
            tags=url_data.tags,
        )
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(e)
        )

    short_url = f"{BASE_URL}/{url['short_code']}"
    response.headers["Location"] = short_url

    return URLResponse(
        short_code=url["short_code"],
        short_url=short_url,
        original_url=url["original_url"],
        created_at=url["created_at"],
        expires_at=url["expires_at"],
        click_count=0,
        tags=url["tags"],
    )

@router.post(
    "/shorten/bulk",
    response_model=List[URLResponse],
    status_code=status.HTTP_201_CREATED,
    summary="Create multiple shortened URLs",
)
def create_bulk_urls(
    bulk: URLBulkCreate,
    db: URLDatabase = Depends(get_database),
):
    """Create multiple shortened URLs in a single request."""
    results = []
    for url_data in bulk.urls:
        try:
            url = db.create_url(
                original_url=url_data.original_url,
                custom_code=url_data.custom_code,
                expires_in_hours=url_data.expires_in_hours,
                tags=url_data.tags,
            )
            results.append(URLResponse(
                short_code=url["short_code"],
                short_url=f"{BASE_URL}/{url['short_code']}",
                original_url=url["original_url"],
                created_at=url["created_at"],
                expires_at=url["expires_at"],
                tags=url["tags"],
            ))
        except ValueError:
            continue  # Skip duplicates in bulk operations
    return results

@router.get(
    "/urls",
    response_model=List[URLResponse],
    summary="List all shortened URLs",
)
def list_urls(
    pagination: PaginationParams = Depends(),
    tag: Optional[str] = None,
    db: URLDatabase = Depends(get_database),
):
    """List all URLs with pagination and optional tag filter."""
    urls = db.list_urls(
        skip=pagination.skip,
        limit=pagination.per_page,
        tag=tag,
    )
    return [
        URLResponse(
            short_code=u["short_code"],
            short_url=f"{BASE_URL}/{u['short_code']}",
            original_url=u["original_url"],
            created_at=u["created_at"],
            expires_at=u["expires_at"],
            click_count=u["click_count"],
            tags=u["tags"],
        )
        for u in urls
    ]

@router.get(
    "/{short_code}",
    summary="Redirect to original URL",
    responses={
        307: {"description": "Redirect to original URL"},
        404: {"description": "Short URL not found"},
    },
)
def redirect_to_url(
    short_code: str,
    response: Response,
    client: dict = Depends(get_client_info),
    visitor_id: Optional[str] = Cookie(None),
    db: URLDatabase = Depends(get_database),
):
    """Redirect to the original URL and track the click."""
    url_data = db.get_url(short_code)
    if not url_data:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Short URL '{short_code}' not found or has expired"
        )

    # Record the click
    db.record_click(
        code=short_code,
        visitor_ip=client["ip"],
        referrer=client["referrer"],
        user_agent=client["user_agent"],
    )

    # Set visitor cookie if not present
    redirect = RedirectResponse(
        url=url_data["original_url"],
        status_code=status.HTTP_307_TEMPORARY_REDIRECT,
    )
    if not visitor_id:
        import uuid
        redirect.set_cookie(
            key="visitor_id",
            value=str(uuid.uuid4()),
            max_age=86400 * 365,
            httponly=True,
        )
    return redirect

@router.get(
    "/{short_code}/info",
    response_model=URLResponse,
    summary="Get URL information without redirecting",
)
def get_url_info(
    short_code: str,
    db: URLDatabase = Depends(get_database),
):
    """Get information about a shortened URL without triggering a redirect."""
    url_data = db.get_url(short_code)
    if not url_data:
        raise HTTPException(status_code=404, detail="URL not found")

    return URLResponse(
        short_code=url_data["short_code"],
        short_url=f"{BASE_URL}/{url_data['short_code']}",
        original_url=url_data["original_url"],
        created_at=url_data["created_at"],
        expires_at=url_data["expires_at"],
        click_count=url_data["click_count"],
        tags=url_data["tags"],
    )

@router.delete(
    "/{short_code}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete a shortened URL",
)
def delete_url(
    short_code: str,
    db: URLDatabase = Depends(get_database),
):
    """Delete a shortened URL."""
    if not db.delete_url(short_code):
        raise HTTPException(status_code=404, detail="URL not found")

14.6 Analytics Routes

# routers/analytics.py
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from typing import Optional
from models import URLStats
from database import URLDatabase
from dependencies import get_database

router = APIRouter(
    prefix="/analytics",
    tags=["Analytics"],
)

@router.get(
    "/{short_code}",
    response_model=URLStats,
    summary="Get URL analytics",
)
def get_url_analytics(
    short_code: str,
    db: URLDatabase = Depends(get_database),
):
    """Get detailed analytics for a shortened URL."""
    stats = db.get_stats(short_code)
    if not stats:
        raise HTTPException(status_code=404, detail="URL not found")
    return stats

@router.get(
    "/{short_code}/export",
    summary="Export click data as CSV",
)
def export_clicks_csv(
    short_code: str,
    db: URLDatabase = Depends(get_database),
):
    """Export click data for a URL as a CSV file using StreamingResponse."""
    url_data = db.get_url(short_code)
    if not url_data:
        raise HTTPException(status_code=404, detail="URL not found")

    clicks = db.clicks.get(short_code, [])

    def generate_csv():
        yield "timestamp,visitor_ip,referrer,user_agent\n"
        for click in clicks:
            yield (
                f"{click['timestamp']},"
                f"{click['visitor_ip']},"
                f"{click['referrer']},"
                f"{click['user_agent']}\n"
            )

    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename={short_code}_clicks.csv"
        },
    )

14.7 Admin Routes

# routers/admin.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from database import URLDatabase
from dependencies import get_database, get_api_key

router = APIRouter(
    prefix="/admin",
    tags=["Admin"],
    dependencies=[Depends(get_api_key)],  # All admin routes require API key
)

@router.get("/dashboard")
def admin_dashboard(db: URLDatabase = Depends(get_database)):
    """Admin dashboard with overview statistics."""
    total_urls = len(db.urls)
    total_clicks = sum(u["click_count"] for u in db.urls.values())
    active_urls = sum(1 for u in db.urls.values() if u["is_active"])

    return {
        "total_urls": total_urls,
        "active_urls": active_urls,
        "total_clicks": total_clicks,
        "avg_clicks_per_url": round(total_clicks / max(total_urls, 1), 2),
    }

@router.delete("/purge-expired")
def purge_expired(db: URLDatabase = Depends(get_database)):
    """Remove all expired URLs."""
    from datetime import datetime
    expired_codes = []
    for code, url_data in list(db.urls.items()):
        if url_data["expires_at"]:
            expires_at = datetime.fromisoformat(url_data["expires_at"])
            if datetime.utcnow() > expires_at:
                expired_codes.append(code)
                db.delete_url(code)

    return {
        "purged_count": len(expired_codes),
        "purged_codes": expired_codes,
    }

14.8 Main Application

# main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from routers import urls, analytics, admin
import time
import uuid

# Create the application
app = FastAPI(
    title="URL Shortener",
    description="A feature-rich URL shortener built with FastAPI",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware for request tracking
class RequestTrackingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        start_time = time.time()

        request.state.request_id = request_id

        response = await call_next(request)

        duration = time.time() - start_time
        response.headers["X-Request-Id"] = request_id
        response.headers["X-Process-Time"] = f"{duration:.4f}s"

        return response

app.add_middleware(RequestTrackingMiddleware)

# Include routers
app.include_router(urls.router)
app.include_router(analytics.router)
app.include_router(admin.router)

# Root endpoint
@app.get("/", tags=["Root"])
def root():
    return {
        "service": "URL Shortener",
        "version": "1.0.0",
        "docs": "/docs",
        "endpoints": {
            "shorten": "POST /shorten",
            "redirect": "GET /{short_code}",
            "info": "GET /{short_code}/info",
            "analytics": "GET /analytics/{short_code}",
            "admin": "GET /admin/dashboard",
        },
    }

# Health check
@app.get("/health", tags=["Root"])
def health_check():
    return {"status": "healthy", "timestamp": time.time()}

# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "detail": "Internal server error",
            "request_id": getattr(request.state, "request_id", "unknown"),
        },
    )

14.9 Testing the URL Shortener

# Start the server
uvicorn main:app --reload

# Create a shortened URL
curl -X POST http://localhost:8000/shorten \
     -H "Content-Type: application/json" \
     -d '{"original_url": "https://www.python.org/doc/", "tags": ["python", "docs"]}'

# Create with custom code
curl -X POST http://localhost:8000/shorten \
     -H "Content-Type: application/json" \
     -d '{"original_url": "https://fastapi.tiangolo.com", "custom_code": "fastapi", "expires_in_hours": 720}'

# Visit the shortened URL (will redirect)
curl -L http://localhost:8000/fastapi

# Get URL info without redirecting
curl http://localhost:8000/fastapi/info

# Get analytics
curl http://localhost:8000/analytics/fastapi

# Export clicks as CSV
curl http://localhost:8000/analytics/fastapi/export -o clicks.csv

# List all URLs with pagination
curl "http://localhost:8000/urls?page=1&per_page=10"

# Filter by tag
curl "http://localhost:8000/urls?tag=python"

# Admin dashboard (requires API key)
curl http://localhost:8000/admin/dashboard \
     -H "X-Api-Key: admin-key-001"

# Bulk create
curl -X POST http://localhost:8000/shorten/bulk \
     -H "Content-Type: application/json" \
     -d '{"urls": [
       {"original_url": "https://python.org"},
       {"original_url": "https://fastapi.tiangolo.com"},
       {"original_url": "https://pydantic.dev"}
     ]}'

# Delete a URL
curl -X DELETE http://localhost:8000/fastapi

# Purge expired URLs (admin)
curl -X DELETE http://localhost:8000/admin/purge-expired \
     -H "X-Api-Key: admin-key-001"

14.10 Requirements File

# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.0
python-multipart==0.0.9
aiofiles==24.1.0
httpx==0.27.0

15. Key Takeaways