FastAPI Tutorial for Beginners (2025): Build Python APIs Fast

FastAPI is the fastest-growing Python web framework — built for building APIs with Python 3.8+ type hints, auto-generated docs, and async support. It's used by Microsoft, Uber, Netflix, and thousands of data science and backend teams. Faster to code than Flask, more focused than Django, and natively async. This tutorial takes you from zero to a production-ready REST API.

What you'll learn

FastAPI vs Flask vs Django

Installation

pip install fastapi uvicorn[standard]

Optionally, use a virtual environment:

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install fastapi uvicorn[standard]

Your First FastAPI App

Create 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}

Run the server:

uvicorn main:app --reload

Visit:

• http://localhost:8000 — your API
• http://localhost:8000/docs — Swagger UI (interactive docs, free)
• http://localhost:8000/redoc — ReDoc docs

The --reload flag restarts the server on file changes during development.

Path Parameters

Path parameters are parts of the URL path captured as variables:

from fastapi import FastAPI

app = FastAPI()

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

@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    # Captures paths like /files/home/user/report.txt
    return {"file_path": file_path}

FastAPI automatically validates the type. Request to /users/abc returns a 422 Unprocessable Entity since abc is not an int.

Predefined path values with Enum

from enum import Enum
from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model": model_name, "message": "Deep Learning FTW!"}
    return {"model": model_name}

Query Parameters

Query parameters are key-value pairs after ? in the URL:

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"name": "Foo"}, {"name": "Bar"}, {"name": "Baz"}]

@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

# Optional query parameter
@app.get("/search/")
def search(q: str | None = None, category: str = "all"):
    results = {"category": category}
    if q:
        results["q"] = q
    return results

Access: http://localhost:8000/items/?skip=0&limit=5

Request Body with Pydantic

Pydantic models define the shape of request and response data. FastAPI validates automatically.

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

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float = Field(gt=0, description="Price must be positive")
    tax: float | None = None

@app.post("/items/")
def create_item(item: Item):
    item_dict = item.model_dump()
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict["price_with_tax"] = price_with_tax
    return item_dict

Test with curl:

curl -X POST http://localhost:8000/items/ \
  -H "Content-Type: application/json" \
  -d '{"name": "Widget", "price": 9.99, "tax": 0.5}'

Pydantic field validators

from pydantic import BaseModel, field_validator

class UserCreate(BaseModel):
    username: str
    email: str
    password: str

    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("Username must be alphanumeric")
        return v

    @field_validator("password")
    @classmethod
    def password_min_length(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

Response Models

Control what data gets returned:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    username: str
    email: str
    password: str

class UserPublic(BaseModel):
    id: int
    username: str
    email: str
    # No password field — never returned

fake_db: list[dict] = []

@app.post("/users/", response_model=UserPublic, status_code=201)
def create_user(user: UserCreate):
    new_user = {"id": len(fake_db) + 1, **user.model_dump()}
    fake_db.append(new_user)
    return new_user  # FastAPI filters to UserPublic fields

The response_model parameter tells FastAPI to strip any extra fields (like password) from the response.

HTTP Status Codes

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {1: "Apple", 2: "Banana"}

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found"
        )
    return {"item": items[item_id]}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    del items[item_id]

Common status codes:

Async Endpoints

FastAPI supports both sync and async endpoints:

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# Sync — runs in a thread pool automatically
@app.get("/sync")
def sync_endpoint():
    return {"type": "sync"}

# Async — runs in the event loop
@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(0.1)  # Non-blocking
    return {"type": "async"}

# Async HTTP request
@app.get("/github/{username}")
async def get_github_user(username: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.github.com/users/{username}")
        return response.json()

When to use async: I/O-bound operations — database queries, HTTP calls, file reads.
When to use sync: CPU-bound operations — image processing, heavy computation.

Dependency Injection

Dependencies are reusable components injected into route functions:

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

app = FastAPI()

# Simple dependency
def get_query_params(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(params: Annotated[dict, Depends(get_query_params)]):
    return params

# Dependency with sub-dependency
def get_db():
    # In real code: yield a database session
    db = {"connected": True}
    try:
        yield db
    finally:
        pass  # Close DB connection here

def get_current_user(db: Annotated[dict, Depends(get_db)]):
    # Validate token, query db
    return {"user_id": 1, "username": "alice"}

@app.get("/profile")
def get_profile(user: Annotated[dict, Depends(get_current_user)]):
    return user

Class-based dependencies

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

class PaginationParams:
    def __init__(self, page: int = 1, per_page: int = 10):
        self.page = page
        self.per_page = per_page
        self.skip = (page - 1) * per_page

@app.get("/posts/")
def list_posts(pagination: Annotated[PaginationParams, Depends()]):
    return {
        "page": pagination.page,
        "per_page": pagination.per_page,
        "skip": pagination.skip,
    }

Complete CRUD REST API

A full example with in-memory storage:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import Annotated
from fastapi import Depends

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

# --- Models ---
class NoteBase(BaseModel):
    title: str
    content: str

class NoteCreate(NoteBase):
    pass

class NoteUpdate(BaseModel):
    title: str | None = None
    content: str | None = None

class Note(NoteBase):
    id: int

# --- In-memory "database" ---
notes_db: dict[int, Note] = {}
counter = 0

def next_id() -> int:
    global counter
    counter += 1
    return counter

# --- Routes ---
@app.get("/notes/", response_model=list[Note])
def list_notes(search: str | None = None):
    notes = list(notes_db.values())
    if search:
        notes = [n for n in notes if search.lower() in n.title.lower()]
    return notes

@app.get("/notes/{note_id}", response_model=Note)
def get_note(note_id: int):
    if note_id not in notes_db:
        raise HTTPException(status_code=404, detail="Note not found")
    return notes_db[note_id]

@app.post("/notes/", response_model=Note, status_code=201)
def create_note(note: NoteCreate):
    new_note = Note(id=next_id(), **note.model_dump())
    notes_db[new_note.id] = new_note
    return new_note

@app.put("/notes/{note_id}", response_model=Note)
def update_note(note_id: int, note: NoteCreate):
    if note_id not in notes_db:
        raise HTTPException(status_code=404, detail="Note not found")
    updated = Note(id=note_id, **note.model_dump())
    notes_db[note_id] = updated
    return updated

@app.patch("/notes/{note_id}", response_model=Note)
def partial_update_note(note_id: int, note: NoteUpdate):
    if note_id not in notes_db:
        raise HTTPException(status_code=404, detail="Note not found")
    existing = notes_db[note_id]
    update_data = note.model_dump(exclude_unset=True)
    updated = existing.model_copy(update=update_data)
    notes_db[note_id] = updated
    return updated

@app.delete("/notes/{note_id}", status_code=204)
def delete_note(note_id: int):
    if note_id not in notes_db:
        raise HTTPException(status_code=404, detail="Note not found")
    del notes_db[note_id]

Test with curl:

# Create
curl -X POST http://localhost:8000/notes/ \
  -H "Content-Type: application/json" \
  -d '{"title": "My note", "content": "Hello world"}'

# List
curl http://localhost:8000/notes/

# Update
curl -X PATCH http://localhost:8000/notes/1 \
  -H "Content-Type: application/json" \
  -d '{"title": "Updated title"}'

# Delete
curl -X DELETE http://localhost:8000/notes/1

JWT Authentication

pip install python-jose[cryptography] passlib[bcrypt]

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# --- Config ---
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Setup ---
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()

# --- Fake user DB ---
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("secret"),
        "email": "alice@example.com",
    }
}

# --- Models ---
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str | None = None

# --- Helpers ---
def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or not verify_password(password, user["hashed_password"]):
        return None
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode["exp"] = expire
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="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:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return User(**user)

# --- Routes ---
@app.post("/token", response_model=Token)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

@app.get("/protected")
async def protected_route(current_user: Annotated[User, Depends(get_current_user)]):
    return {"message": f"Hello, {current_user.username}!"}

Test:

# Login
curl -X POST http://localhost:8000/token \
  -d "username=alice&password=secret"

# Use token
curl http://localhost:8000/users/me \
  -H "Authorization: Bearer <your-token>"

Database with SQLAlchemy (Async)

pip install sqlalchemy asyncpg

from contextlib import asynccontextmanager
from typing import AsyncGenerator, Annotated

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# --- Database setup ---
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# --- Model ---
class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str]

# --- Lifespan (create tables on startup) ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

# --- Dependency ---
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

DB = Annotated[AsyncSession, Depends(get_db)]

# --- Pydantic schemas ---
class PostCreate(BaseModel):
    title: str
    content: str

class PostResponse(BaseModel):
    id: int
    title: str
    content: str
    model_config = {"from_attributes": True}

# --- Routes ---
@app.get("/posts/", response_model=list[PostResponse])
async def list_posts(db: DB):
    result = await db.execute(select(Post))
    return result.scalars().all()

@app.get("/posts/{post_id}", response_model=PostResponse)
async def get_post(post_id: int, db: DB):
    post = await db.get(Post, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

@app.post("/posts/", response_model=PostResponse, status_code=201)
async def create_post(data: PostCreate, db: DB):
    post = Post(**data.model_dump())
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

@app.delete("/posts/{post_id}", status_code=204)
async def delete_post(post_id: int, db: DB):
    post = await db.get(Post, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    await db.delete(post)
    await db.commit()

Environment Variables

pip install python-dotenv

Create .env:

DATABASE_URL=postgresql+asyncpg://user:password@localhost/mydb
SECRET_KEY=supersecretkey
DEBUG=false

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool = False
    app_name: str = "My API"

    model_config = {"env_file": ".env"}

settings = Settings()

pip install pydantic-settings

Use in your app:

from fastapi import FastAPI
from .config import settings

app = FastAPI(title=settings.app_name, debug=settings.debug)

Middleware

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

app = FastAPI()

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

# Custom middleware — request timing
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Background Tasks

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

def send_email(email: str, message: str):
    # Runs after response is sent
    print(f"Sending email to {email}: {message}")

class Item(BaseModel):
    name: str
    owner_email: str

@app.post("/items/")
def create_item(item: Item, background_tasks: BackgroundTasks):
    background_tasks.add_task(
        send_email,
        item.owner_email,
        f"Your item '{item.name}' was created!"
    )
    return {"message": "Item created, email will be sent shortly"}

File Uploads

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse
import shutil
from pathlib import Path

app = FastAPI()

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

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # Validate file type
    allowed_types = {"image/jpeg", "image/png", "image/webp"}
    if file.content_type not in allowed_types:
        return JSONResponse(
            status_code=400,
            content={"detail": "Only JPEG, PNG, WebP allowed"}
        )
    
    file_path = UPLOAD_DIR / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename, "size": file_path.stat().st_size}

@app.post("/upload-multiple/")
async def upload_multiple(files: list[UploadFile] = File(...)):
    results = []
    for file in files:
        file_path = UPLOAD_DIR / file.filename
        with file_path.open("wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        results.append({"filename": file.filename})
    return results

Project Structure

For larger applications:

myapp/
├── main.py              # App entry point
├── config.py            # Settings (pydantic-settings)
├── database.py          # DB engine + session
├── models/              # SQLAlchemy ORM models
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── schemas/             # Pydantic request/response models
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── routers/             # Route handlers
│   ├── __init__.py
│   ├── users.py
│   └── posts.py
├── dependencies/        # Shared dependencies
│   ├── __init__.py
│   └── auth.py
├── services/            # Business logic
│   └── email.py
├── tests/
│   └── test_posts.py
├── .env
└── requirements.txt

main.py:

from fastapi import FastAPI
from .routers import users, posts
from .config import settings

app = FastAPI(title=settings.app_name)

app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(posts.router, prefix="/posts", tags=["posts"])

routers/posts.py:

from fastapi import APIRouter, Depends, HTTPException
from typing import Annotated
from ..dependencies.auth import get_current_user
from ..schemas.post import PostCreate, PostResponse
from ..database import DB

router = APIRouter()

@router.get("/", response_model=list[PostResponse])
async def list_posts(db: DB):
    ...

@router.post("/", response_model=PostResponse, status_code=201)
async def create_post(
    data: PostCreate,
    db: DB,
    current_user: Annotated[dict, Depends(get_current_user)]
):
    ...

Testing

pip install pytest httpx pytest-asyncio

# tests/test_notes.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.mark.anyio
async def test_create_note():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.post(
            "/notes/",
            json={"title": "Test", "content": "Hello"}
        )
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test"
    assert "id" in data

@pytest.mark.anyio
async def test_get_nonexistent_note():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.get("/notes/9999")
    assert response.status_code == 404

@pytest.mark.anyio
async def test_list_notes():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.get("/notes/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

Run:

pip install anyio[trio]
pytest tests/ -v

Deployment

Docker

FROM python:3.12-slim

WORKDIR /app

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

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db/mydb
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mydb
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

Deployment platforms

3 Projects to Build

Project 1 — URL Shortener API

Build a URL shortening service (like bit.ly):

import string
import random
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, HttpUrl

app = FastAPI(title="URL Shortener")

url_db: dict[str, str] = {}

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

class URLCreate(BaseModel):
    url: HttpUrl
    custom_code: str | None = None

class URLResponse(BaseModel):
    short_code: str
    short_url: str
    original_url: str

@app.post("/shorten", response_model=URLResponse, status_code=201)
def shorten_url(data: URLCreate):
    code = data.custom_code or generate_code()
    if code in url_db:
        raise HTTPException(status_code=409, detail="Code already taken")
    url_db[code] = str(data.url)
    return {
        "short_code": code,
        "short_url": f"http://localhost:8000/{code}",
        "original_url": str(data.url),
    }

@app.get("/{code}")
def redirect(code: str):
    if code not in url_db:
        raise HTTPException(status_code=404, detail="Short URL not found")
    return RedirectResponse(url=url_db[code])

@app.get("/stats/{code}")
def get_stats(code: str):
    if code not in url_db:
        raise HTTPException(status_code=404, detail="Not found")
    return {"code": code, "original_url": url_db[code]}

Project 2 — Todo API with Authentication

Full CRUD todo list with user accounts:

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

SECRET_KEY = "change-me"
ALGORITHM = "HS256"

app = FastAPI(title="Todo API with Auth")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

# In-memory stores
users_db: dict[str, dict] = {}
todos_db: dict[int, dict] = {}
todo_counter = 0

class UserRegister(BaseModel):
    username: str
    password: str

class TodoCreate(BaseModel):
    title: str
    done: bool = False

class Todo(BaseModel):
    id: int
    title: str
    done: bool
    owner: str

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username or username not in users_db:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

CurrentUser = Annotated[str, Depends(get_current_user)]

@app.post("/register", status_code=201)
def register(data: UserRegister):
    if data.username in users_db:
        raise HTTPException(status_code=409, detail="Username taken")
    users_db[data.username] = {"hashed_password": pwd_context.hash(data.password)}
    return {"message": "Registered"}

@app.post("/login")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = users_db.get(form.username)
    if not user or not pwd_context.verify(form.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Bad credentials")
    token = jwt.encode(
        {"sub": form.username, "exp": datetime.now(timezone.utc) + timedelta(hours=24)},
        SECRET_KEY, algorithm=ALGORITHM
    )
    return {"access_token": token, "token_type": "bearer"}

@app.get("/todos", response_model=list[Todo])
def list_todos(user: CurrentUser):
    return [t for t in todos_db.values() if t["owner"] == user]

@app.post("/todos", response_model=Todo, status_code=201)
def create_todo(data: TodoCreate, user: CurrentUser):
    global todo_counter
    todo_counter += 1
    todo = {"id": todo_counter, **data.model_dump(), "owner": user}
    todos_db[todo_counter] = todo
    return todo

@app.patch("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, data: TodoCreate, user: CurrentUser):
    todo = todos_db.get(todo_id)
    if not todo or todo["owner"] != user:
        raise HTTPException(status_code=404, detail="Not found")
    todos_db[todo_id].update(data.model_dump())
    return todos_db[todo_id]

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, user: CurrentUser):
    todo = todos_db.get(todo_id)
    if not todo or todo["owner"] != user:
        raise HTTPException(status_code=404, detail="Not found")
    del todos_db[todo_id]

Project 3 — ML Model Serving API

Serve a trained scikit-learn model:

pip install scikit-learn joblib

# train_model.py — run once to save model
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import joblib

iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    iris.data, iris.target, test_size=0.2, random_state=42
)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
joblib.dump(model, "iris_model.pkl")
print(f"Accuracy: {model.score(X_test, y_test):.2f}")

# main.py
from contextlib import asynccontextmanager
import joblib
import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel, Field

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    ml_models["iris"] = joblib.load("iris_model.pkl")
    yield
    ml_models.clear()

app = FastAPI(title="Iris Classifier API", lifespan=lifespan)

CLASSES = ["setosa", "versicolor", "virginica"]

class IrisFeatures(BaseModel):
    sepal_length: float = Field(..., gt=0, example=5.1)
    sepal_width: float = Field(..., gt=0, example=3.5)
    petal_length: float = Field(..., gt=0, example=1.4)
    petal_width: float = Field(..., gt=0, example=0.2)

class Prediction(BaseModel):
    class_id: int
    class_name: str
    probability: float

@app.post("/predict", response_model=Prediction)
def predict(features: IrisFeatures):
    X = np.array([[
        features.sepal_length,
        features.sepal_width,
        features.petal_length,
        features.petal_width,
    ]])
    model = ml_models["iris"]
    class_id = int(model.predict(X)[0])
    probability = float(model.predict_proba(X)[0][class_id])
    return {
        "class_id": class_id,
        "class_name": CLASSES[class_id],
        "probability": probability,
    }

@app.get("/health")
def health():
    return {"status": "ok", "model_loaded": "iris" in ml_models}

Test:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}'
# → {"class_id": 0, "class_name": "setosa", "probability": 0.97}

Common Mistakes

FastAPI vs Related Terms

Learning Path

Free Resources

Quick Reference

from fastapi import FastAPI, Depends, HTTPException, status, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

# Middleware
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# Pydantic model
class Item(BaseModel):
    name: str
    price: float

# Dependency
def get_db():
    yield {"connected": True}

DB = Annotated[dict, Depends(get_db)]

# Routes
@app.get("/")                      # GET
@app.post("/", status_code=201)   # POST → 201 Created
@app.put("/{id}")                  # PUT (full replace)
@app.patch("/{id}")                # PATCH (partial update)
@app.delete("/{id}", status_code=204)  # DELETE → 204 No Content

# Raise HTTP errors
raise HTTPException(status_code=404, detail="Not found")
raise HTTPException(status_code=401, headers={"WWW-Authenticate": "Bearer"})

# Background task
def send_email(address: str): ...
background_tasks.add_task(send_email, "user@example.com")

# Run
# uvicorn main:app --reload

FAQ

Do I need to know Flask before FastAPI?
No. FastAPI is beginner-friendly. Basic Python knowledge (functions, classes, type hints) is sufficient. Django/Flask experience helps but isn't required.

FastAPI vs Django REST Framework?
FastAPI is faster, has automatic docs, and native async support. DRF is more mature with admin, full auth, and more built-in tools. Use FastAPI for greenfield APIs and ML serving; DRF/Django when you need the full batteries-included ecosystem.

Is FastAPI production-ready?
Yes. Used by Uber, Netflix, Microsoft, and hundreds of companies in production. Active development, stable API since 0.95+.

Do I need to use async for everything?
No. Regular def routes work fine — FastAPI runs them in a thread pool automatically. Use async def when you have I/O-bound operations (DB queries, HTTP calls) that support async.

What database should I use with FastAPI?
PostgreSQL with SQLAlchemy async is the most common combo. MongoDB with Motor, Redis with aioredis, and SQLite with aiosqlite are also popular. For simple apps, SQLite is fine.

How do I add API versioning?
Use router prefixes: app.include_router(v1_router, prefix="/api/v1") and app.include_router(v2_router, prefix="/api/v2"). Or use separate FastAPI apps mounted on different paths.