FastAPI Tutorial — From Zero to Production with Diagrams and Code

The Story That Explains FastAPI

FastAPI is a modern, high-performance web framework for building APIs with Python 3.8+ based on standard type hints. It sits on top of Starlette (for the web parts) and Pydantic (for the data parts), giving you async speed, automatic validation, and auto-generated interactive docs — all from the same function signature you'd write anyway.

Installation and Your First API

You need Python 3.8+ and two packages: fastapi for the framework and uvicorn as the ASGI server that runs it in production.

# Install FastAPI and the Uvicorn ASGI server
pip install "fastapi[standard]"

# Or minimal install
pip install fastapi uvicorn

Hello World — The Shortest Possible API

# file: main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

Now run it:

# Development mode with auto-reload
uvicorn main:app --reload

# Or with the new fastapi CLI
fastapi dev main.py

The Three Pillars of Every Endpoint

Every FastAPI endpoint deals with three sources of incoming data. Understanding these three is 80% of understanding FastAPI.

Type Hints — The Heart of FastAPI

The Same Function — Progressively Typed

# Level 0 — No hints. Everything is a string. Errors happen inside your function.
@app.get("/items/{item_id}")
def get_item(item_id):
    return {"id": item_id}

# Level 1 — Add a type hint. FastAPI now validates + converts.
@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"id": item_id}

# Level 2 — Add constraints with Path and Query.
from fastapi import Path, Query

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(..., gt=0, le=1000),
    q:       str | None = Query(None, max_length=50),
):
    return {"id": item_id, "q": q}

Visual Diagram — Request Lifecycle

Every HTTP request travels through the same pipeline. Knowing each stage helps you plug in middleware, dependencies, and error handling at the right place.

HTTP Methods — CRUD in FastAPI

A REST API maps standard HTTP verbs to database operations. FastAPI has a decorator for each verb — the mapping is one-to-one and enforced.

Complete CRUD API — In-Memory Item Store

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from uuid import UUID, uuid4

app = FastAPI(title="Item API", version="1.0.0")

# ── Models ────────────────────────────────────────────
class ItemIn(BaseModel):
    name:  str   = Field(..., min_length=1, max_length=80)
    price: float = Field(..., gt=0)
    stock: int   = Field(0, ge=0)

class ItemOut(ItemIn):
    id: UUID

# ── Fake in-memory DB ─────────────────────────────────
DB: dict[UUID, ItemOut] = {}

# ── CREATE ────────────────────────────────────────────
@app.post("/items", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
def create_item(payload: ItemIn):
    new = ItemOut(id=uuid4(), **payload.model_dump())
    DB[new.id] = new
    return new

# ── READ (list) ───────────────────────────────────────
@app.get("/items", response_model=list[ItemOut])
def list_items(skip: int = 0, limit: int = 20):
    return list(DB.values())[skip : skip + limit]

# ── READ (one) ────────────────────────────────────────
@app.get("/items/{item_id}", response_model=ItemOut)
def get_item(item_id: UUID):
    if item_id not in DB:
        raise HTTPException(404, "Item not found")
    return DB[item_id]

# ── UPDATE ────────────────────────────────────────────
@app.put("/items/{item_id}", response_model=ItemOut)
def update_item(item_id: UUID, payload: ItemIn):
    if item_id not in DB:
        raise HTTPException(404, "Item not found")
    DB[item_id] = ItemOut(id=item_id, **payload.model_dump())
    return DB[item_id]

# ── DELETE ────────────────────────────────────────────
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: UUID):
    if DB.pop(item_id, None) is None:
        raise HTTPException(404, "Item not found")

Path and Query Parameters — Deep Dive

Path Parameter Validation

from fastapi import FastAPI, Path
from enum import Enum

class Category(str, Enum):
    electronics = "electronics"
    books       = "books"
    clothing    = "clothing"

@app.get("/products/{category}/{product_id}")
def get_product(
    category:   Category,                                    # enum → only 3 allowed values
    product_id: int = Path(..., ge=1, le=999999,        # must be 1..999999
                            title="Product ID",
                            description="Unique product identifier"),
):
    return {"category": category, "product_id": product_id}

Query Parameter Patterns

from fastapi import Query
from typing import Annotated

@app.get("/search")
def search(
    # Required query param
    q:       Annotated[str,  Query(min_length=3, max_length=50)],
    # Optional with default
    limit:   Annotated[int,  Query(ge=1, le=100)] = 10,
    # Boolean flag
    exact:   Annotated[bool, Query()] = False,
    # List — accept ?tag=a&tag=b&tag=c
    tag:     Annotated[list[str] | None, Query()] = None,
    # Regex validation
    sort_by: Annotated[str, Query(pattern="^(name|price|date)$")] = "name",
):
    return {"q": q, "limit": limit, "exact": exact,
            "tag": tag or [], "sort_by": sort_by}

Request Body with Pydantic Models

Pydantic is FastAPI's data-validation engine. Declare a class inheriting from BaseModel, and every incoming JSON payload gets parsed, validated, and transformed into a typed Python object automatically.

from pydantic import BaseModel, Field, EmailStr, HttpUrl, field_validator
from datetime import date
from typing import Annotated

class Address(BaseModel):
    street:      str
    city:        str
    postal_code: str = Field(..., pattern=r"^\d{5,6}$")
    country:     str = "IN"

class UserRegistration(BaseModel):
    username:    str      = Field(..., min_length=3, max_length=20)
    email:       EmailStr
    password:    str      = Field(..., min_length=8)
    birth_date:  date
    website:     HttpUrl | None = None
    address:     Address                         # nested model
    tags:        list[str] = []

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password needs at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password needs at least one digit")
        return v

@app.post("/register")
def register(user: UserRegistration):
    # user is fully validated by the time we get here
    return {"status": "created", "user": user.model_dump(exclude={"password"})}

What Happens on Invalid Input

Response Models and Status Codes

Declaring a response_model gives you three benefits at once: automatic output validation, automatic filtering of sensitive fields, and rich auto-generated documentation.

from fastapi import FastAPI, status
from pydantic import BaseModel, EmailStr

# Two versions: one with the password, one without
class UserIn(BaseModel):
    username: str
    email:    EmailStr
    password: str                # client sends this

class UserOut(BaseModel):
    username: str
    email:    EmailStr           # NO password field

@app.post("/users",
          response_model=UserOut,
          status_code=status.HTTP_201_CREATED,
          summary="Register a new user",
          tags=["users"])
def create_user(user: UserIn) -> UserIn:
    # Return the whole UserIn (with password) — FastAPI filters it
    # out because response_model=UserOut has no password field.
    return user

Dependency Injection — FastAPI's Secret Weapon

from fastapi import Depends, HTTPException, Header
from typing import Annotated

# ── A simple dependency ───────────────────────────────
def get_pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": min(limit, 100)}

# ── A dependency with sub-dependencies ───────────────
def verify_api_key(x_api_key: Annotated[str, Header()]):
    if x_api_key != "secret-key-123":
        raise HTTPException(401, "Invalid API key")
    return x_api_key

def get_current_user(api_key: Annotated[str, Depends(verify_api_key)]):
    # Real app: look up user by api_key in DB
    return {"id": "u_42", "name": "Alice"}

# ── Yield-based dependency (context manager style) ───
def get_db():
    db = SessionLocal()
    try:
        yield db          # pass to endpoint
    finally:
        db.close()        # always cleanup, even on error

# ── Consuming all three ──────────────────────────────
@app.get("/orders")
def list_orders(
    pagination: Annotated[dict, Depends(get_pagination)],
    user:       Annotated[dict, Depends(get_current_user)],
    db:         Annotated[Session, Depends(get_db)],
):
    return {"user": user, "page": pagination, "orders": []}

Authentication with JWT (OAuth2)

FastAPI ships with OAuth2 primitives built-in. Combined with the python-jose library for JWT and passlib for password hashing, you get production-grade auth in about 60 lines.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from typing import Annotated

SECRET_KEY = "replace-me-with-openssl-rand-hex-32"
ALGORITHM  = "HS256"
TOKEN_TTL  = timedelta(minutes=30)

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2  = OAuth2PasswordBearer(tokenUrl="/token")

# Fake user DB — real app: query PostgreSQL etc.
USERS = {
    "alice": {"username": "alice",
              "hashed_pw": pwd_ctx.hash("wonderland")},
}

def authenticate(username: str, password: str):
    user = USERS.get(username)
    if not user or not pwd_ctx.verify(password, user["hashed_pw"]):
        return None
    return user

def create_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.now(timezone.utc) + TOKEN_TTL
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(401, "Wrong username or password")
    token = create_token({"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}

def get_current_user(token: Annotated[str, Depends(oauth2)]):
    creds_exc = HTTPException(401, "Could not validate credentials",
                              headers={"WWW-Authenticate": "Bearer"})
    try:
        payload  = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None or username not in USERS:
            raise creds_exc
    except JWTError:
        raise creds_exc
    return USERS[username]

@app.get("/me")
def read_me(user: Annotated[dict, Depends(get_current_user)]):
    return {"username": user["username"]}

Database Integration with SQLAlchemy

# file: database.py
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine       = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base         = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id    = Column(Integer, primary_key=True, index=True)
    name  = Column(String, nullable=False)
    price = Column(Float,  nullable=False)
    stock = Column(Integer, default=0)

Base.metadata.create_all(bind=engine)

# file: main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from database import SessionLocal, Item

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:     yield db
    finally: db.close()

class ItemSchema(BaseModel):
    name:  str
    price: float
    stock: int = 0
    class Config:
        from_attributes = True    # read ORM objects directly

@app.post("/items", response_model=ItemSchema, status_code=201)
def create(payload: ItemSchema, db: Session = Depends(get_db)):
    item = Item(**payload.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

@app.get("/items/{item_id}", response_model=ItemSchema)
def read(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(404, "Item not found")
    return item

Middleware and CORS

Middleware wraps every request — perfect for logging, timing, CORS, GZip compression, and trace propagation. Register in order; they execute in that order on the way in and in reverse order on the way out.

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

app = FastAPI()

# ── CORS — allow the browser to call your API ───────
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── GZip — compress large responses automatically ───
app.add_middleware(GZipMiddleware, minimum_size=1000)

# ── Custom timing middleware ─────────────────────────
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start    = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response

Sync vs Async — When Speed Matters

# ── SYNC — blocks the worker while waiting ──────────
import requests

@app.get("/weather-sync/{city}")
def weather_sync(city: str):
    r = requests.get(f"https://api.weather.com/{city}", timeout=5)
    return r.json()

# ── ASYNC — releases the worker while waiting ───────
import httpx

@app.get("/weather-async/{city}")
async def weather_async(city: str):
    async with httpx.AsyncClient(timeout=5) as client:
        r = await client.get(f"https://api.weather.com/{city}")
    return r.json()

Testing FastAPI Applications

FastAPI includes TestClient, a wrapper around httpx that lets you write synchronous tests without spinning up a real server. Pair it with pytest for a fast, reliable test loop.

# file: test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_item():
    payload = {"name": "Notebook", "price": 2.99, "stock": 100}
    r = client.post("/items", json=payload)
    assert r.status_code == 201
    data = r.json()
    assert data["name"]  == "Notebook"
    assert data["price"] == 2.99
    assert "id" in data

def test_create_item_validation_error():
    payload = {"name": "", "price": -1}    # invalid
    r = client.post("/items", json=payload)
    assert r.status_code == 422
    errors = r.json()["detail"]
    assert len(errors) >= 2            # name + price both wrong

def test_get_missing_item_returns_404():
    r = client.get("/items/00000000-0000-0000-0000-000000000000")
    assert r.status_code == 404
    assert r.json()["detail"] == "Item not found"

FastAPI vs Flask vs Django REST

Deployment — From Laptop to Production

Production Server with Gunicorn + Uvicorn Workers

# Install
pip install "fastapi[standard]" gunicorn uvicorn

# Run with 4 workers
gunicorn main:app \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    -b 0.0.0.0:8000 \
    --access-logfile - \
    --error-logfile -

Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "main:app",
     "-w", "4",
     "-k", "uvicorn.workers.UvicornWorker",
     "-b", "0.0.0.0:8000"]

When to Use FastAPI

Golden Rules