FastAPI Tutorial — Build REST APIs with Python (Complete Guide)

Why FastAPI?

FastAPI gives you three things simultaneously that used to require trade-offs:

• Performance: On par with Node.js/Go for I/O-bound workloads (built on Starlette + uvicorn)
• Developer experience: Type hints drive automatic validation, serialization, and interactive docs
• Production-ready: Built-in OAuth2, async, WebSockets, background tasks, and OpenAPI out of the box

Installation

pip install fastapi uvicorn[standard]

# Or with all optional extras
pip install "fastapi[all]"

Your First API

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="Example FastAPI application",
    version="1.0.0",
)

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

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

# Run the server
uvicorn main:app --reload

# API docs available at:
# http://localhost:8000/docs  (Swagger UI)
# http://localhost:8000/redoc (ReDoc)

Request and Response Models with Pydantic

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, field_validator
from datetime import datetime
from typing import Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: int
    bio: Optional[str] = None

    @field_validator("age")
    @classmethod
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError("Age must be between 0 and 150")
        return v

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

    model_config = {"from_attributes": True}  # enables ORM mode

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # FastAPI validates the request body against UserCreate
    # and serializes the response with UserResponse
    db_user = await save_user_to_db(user)  # your DB logic
    return db_user

Dependency Injection

FastAPI's dependency system cleanly handles cross-cutting concerns like auth, database sessions, and rate limiting:

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

# Database session dependency
async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

# Authentication dependency
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
):
    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])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None:
        raise credentials_exception
    return user

# Use in route -- automatically handles auth + DB session
@app.get("/profile", response_model=UserResponse)
async def get_profile(
    current_user: User = Depends(get_current_user)
):
    return current_user

Async Database Access with SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

# CRUD operations
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Error Handling

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

app = FastAPI()

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Exception handler
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": "not_found", "message": f"Item {exc.item_id} not found"},
    )

# Generic 500 handler
@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # Log the error here
    return JSONResponse(
        status_code=500,
        content={"error": "internal_error", "message": "An unexpected error occurred"},
    )

Middleware

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

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

@app.middleware("http")
async def add_request_timing(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:.3f}"
    return response

Background Tasks

from fastapi import BackgroundTasks

def send_welcome_email(user_email: str, user_name: str):
    # This runs after the response is sent
    send_email(user_email, f"Welcome, {user_name}!")

@app.post("/users", response_model=UserResponse)
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
):
    db_user = await save_user(db, user)
    background_tasks.add_task(
        send_welcome_email, user.email, user.name
    )
    return db_user  # Returns immediately; email sends in background

Production Deployment

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

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

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:pass@db/mydb
    depends_on:
      - db
  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb

Frequently Asked Questions

FastAPI vs Flask: which should I choose?

Choose FastAPI for new projects: async-native, faster, automatic validation/docs, better type safety. Choose Flask if you're extending an existing Flask app, need a specific Flask extension, or your team already knows Flask well.

Can FastAPI replace Django?

For pure APIs, yes. For full-stack apps with Django admin, ORM migrations (Alembic covers this for FastAPI), auth systems, and templates, Django is still more batteries-included. Many teams use FastAPI for the API layer and keep Django Admin for internal tools.

How do I handle file uploads?

Use UploadFile and File from fastapi: async def upload(file: UploadFile = File(...)). Read the file with await file.read() or stream it with file.file.

Does FastAPI support WebSockets?

Yes, natively. Use @app.websocket("/ws") decorator with async def ws_endpoint(websocket: WebSocket). The WebSocket object has accept(), receive_text(), send_text(), and close() methods.