Python: Zero to Hero
Home/The Python Ecosystem
Share

Chapter 41: Web Development with FastAPI

FastAPI is the fastest-growing Python web framework. It's built for APIs, runs on async Python, validates data automatically using type hints, and generates interactive documentation for free.

Companies like Uber, Netflix, and Microsoft use it. If you're building an API in 2026, FastAPI is the default choice.

What Makes FastAPI Different?

Feature Flask FastAPI
Speed Fast Very fast (Starlette + async)
Data validation Manual Automatic (Pydantic)
API docs Manual Auto-generated (Swagger + ReDoc)
Type hints Optional Built-in, used for validation
Async support Limited First-class
Learning curve Lower Slightly higher

You already know Flask from Chapter 40. FastAPI feels familiar but more powerful.

Setup

pip install fastapi uvicorn[standard] pydantic

uvicorn is the ASGI server that runs FastAPI. Think of it like gunicorn for Flask, but async.

Your First FastAPI App

# main.py
from fastapi import FastAPI

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


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


@app.get("/about")
def about():
    return {"app": "My API", "author": "Alice"}

Run it:

uvicorn main:app --reload

Open http://127.0.0.1:8000/ — you'll see the JSON response.

Now open http://127.0.0.1:8000/docs — you get a full interactive Swagger UI. Every endpoint is documented, testable, and shows request/response schemas. No extra code required.

--reload restarts the server on code changes, like debug=True in Flask.

Path Parameters

@app.get("/users/{user_id}")
def get_user(user_id: int):
    # FastAPI validates that user_id is an int
    # If you pass /users/abc -> automatic 422 error
    return {"user_id": user_id}


@app.get("/items/{item_name}")
def get_item(item_name: str):
    return {"item": item_name}

The type hint user_id: int does two things at once:

  1. Converts the string from the URL to an integer
  2. Returns a 422 Unprocessable Entity error if it's not a valid integer

Query Parameters

from typing import Optional


@app.get("/search")
def search(
    q:        str,
    page:     int = 1,
    per_page: int = 10,
    active:   Optional[bool] = None,
):
    # /search?q=python
    # /search?q=python&page=2&per_page=20
    # /search?q=python&active=true
    return {
        "query":    q,
        "page":     page,
        "per_page": per_page,
        "active":   active,
    }

Parameters with default values are optional query params. Parameters without defaults are required.

Request Bodies with Pydantic

This is where FastAPI gets powerful. Define a Pydantic model and FastAPI uses it for:

  • Automatic JSON parsing
  • Validation
  • Error messages
  • Documentation
from pydantic import BaseModel, Field, EmailStr
from typing import Optional


class UserCreate(BaseModel):
    name:     str             = Field(..., min_length=1, max_length=100)
    email:    str             = Field(..., description="User's email address")
    age:      int             = Field(..., ge=0, le=150)
    bio:      Optional[str]   = None


@app.post("/users", status_code=201)
def create_user(user: UserCreate):
    # user is already validated — if we get here, it's clean data
    return {
        "id":    42,
        "name":  user.name,
        "email": user.email,
        "age":   user.age,
    }

If you send this:

{"name": "", "email": "not-an-email", "age": -5}

You get back:

{
  "detail": [
    {"loc": ["body", "name"], "msg": "ensure this value has at least 1 character"},
    {"loc": ["body", "age"],  "msg": "ensure this value is greater than or equal to 0"}
  ]
}

Zero code written for validation. Pydantic handles everything.

Field validators

from pydantic import BaseModel, field_validator
import re


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

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

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

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email address")
        return v.lower()

Response Models

Define what the API returns separately from what it accepts. This lets you hide sensitive fields (like passwords) from responses:

from pydantic import BaseModel
from datetime import datetime


class UserCreate(BaseModel):
    name:     str
    email:    str
    password: str   # accepted in request


class UserOut(BaseModel):
    id:         int
    name:       str
    email:      str
    created_at: datetime
    # password is NOT here — never returned


@app.post("/users", response_model=UserOut, status_code=201)
def create_user(user: UserCreate):
    # In a real app: hash password, save to DB
    return UserOut(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.utcnow(),
    )

FastAPI filters the response through UserOut automatically — even if you return extra fields, they won't appear in the output.

A Full CRUD API

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

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


# ── Models ────────────────────────────────────────────────────────────────────

class TaskCreate(BaseModel):
    title:       str            = Field(..., min_length=1, max_length=200)
    description: Optional[str]  = None
    priority:    int            = Field(1, ge=1, le=5)


class TaskUpdate(BaseModel):
    title:       Optional[str]  = Field(None, min_length=1, max_length=200)
    description: Optional[str]  = None
    priority:    Optional[int]  = Field(None, ge=1, le=5)
    done:        Optional[bool] = None


class TaskOut(BaseModel):
    id:          int
    title:       str
    description: Optional[str]
    priority:    int
    done:        bool
    created_at:  datetime


# ── In-memory store ───────────────────────────────────────────────────────────

tasks:   dict[int, TaskOut] = {}
next_id: int = 1


# ── Endpoints ─────────────────────────────────────────────────────────────────

@app.get("/tasks", response_model=list[TaskOut])
def list_tasks(
    done:     Optional[bool] = Query(None, description="Filter by done status"),
    priority: Optional[int]  = Query(None, ge=1, le=5),
    limit:    int             = Query(20, ge=1, le=100),
    offset:   int             = Query(0, ge=0),
):
    result = list(tasks.values())
    if done is not None:
        result = [t for t in result if t.done == done]
    if priority is not None:
        result = [t for t in result if t.priority == priority]
    return result[offset : offset + limit]


@app.post("/tasks", response_model=TaskOut, status_code=201)
def create_task(task: TaskCreate):
    global next_id
    new_task = TaskOut(
        id=next_id,
        title=task.title,
        description=task.description,
        priority=task.priority,
        done=False,
        created_at=datetime.utcnow(),
    )
    tasks[next_id] = new_task
    next_id += 1
    return new_task


@app.get("/tasks/{task_id}", response_model=TaskOut)
def get_task(task_id: int):
    task = tasks.get(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@app.patch("/tasks/{task_id}", response_model=TaskOut)
def update_task(task_id: int, updates: TaskUpdate):
    task = tasks.get(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")

    task_data = task.model_dump()
    update_data = updates.model_dump(exclude_unset=True)  # only changed fields
    task_data.update(update_data)
    tasks[task_id] = TaskOut(**task_data)
    return tasks[task_id]


@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    del tasks[task_id]

Visit http://127.0.0.1:8000/docs to test every endpoint interactively.

Async Endpoints

FastAPI is built on asyncio. Use async def for I/O-bound routes:

import asyncio
import httpx   # async HTTP client


@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()

    return {
        "city":        city,
        "temp_c":      data["current_condition"][0]["temp_C"],
        "description": data["current_condition"][0]["weatherDesc"][0]["value"],
    }

pip install httpx

Use async def when your endpoint does I/O (database queries, HTTP calls, file reads). Use regular def for CPU-bound or synchronous code. FastAPI handles both correctly.

Dependency Injection

Dependencies let you share logic across endpoints — database connections, authentication, pagination params:

from fastapi import Depends


# ── A reusable "pagination" dependency ────────────────────────────────────────

class Pagination:
    def __init__(self, limit: int = 20, offset: int = 0):
        self.limit  = limit
        self.offset = offset


@app.get("/products")
def list_products(pagination: Pagination = Depends()):
    # pagination.limit and pagination.offset are set automatically
    return {"limit": pagination.limit, "offset": pagination.offset}


# ── A reusable "current user" dependency ──────────────────────────────────────

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()


def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    token = credentials.credentials
    # In a real app: decode JWT, look up user in DB
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"id": 1, "name": "Alice"}


@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user


@app.get("/dashboard")
def dashboard(current_user: dict = Depends(get_current_user)):
    return {"welcome": current_user["name"]}

Both /me and /dashboard are now protected without repeating the auth logic.

Authentication with JWT

A realistic authentication flow:

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

from datetime import datetime, timedelta
from typing import Optional
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

SECRET_KEY  = "super-secret-key-change-this"
ALGORITHM   = "HS256"
TOKEN_EXPIRY = timedelta(hours=24)

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

# Fake user database
USERS_DB = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("password123"),
        "email": "alice@example.com",
    }
}


class Token(BaseModel):
    access_token: str
    token_type:   str


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


def create_access_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + TOKEN_EXPIRY
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload  = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = USERS_DB.get(username)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    return user


@app.post("/token", response_model=Token)
def login(form: 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=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token({"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}


@app.get("/me", response_model=UserOut)
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Test it in /docs — click "Authorize," log in as alice/password123, then call /me.

Lifespan Events — Startup and Shutdown

Use lifespan to run code once at startup and shutdown (connect to DB, warm up caches):

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up — connecting to database...")
    app.state.db = {}   # replace with real DB connection

    yield   # app is running here

    # Shutdown
    print("Shutting down — closing connections...")


app = FastAPI(lifespan=lifespan)


@app.get("/")
def home():
    return {"status": "running"}

Middleware

Middleware runs before and after every request:

import time
from fastapi import Request


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

Add CORS support (required for browser-based frontends):

from fastapi.middleware.cors import CORSMiddleware

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

Background Tasks

Run slow tasks (sending emails, processing files) after returning the response:

from fastapi import BackgroundTasks


def send_welcome_email(email: str, name: str):
    import time
    time.sleep(2)   # simulate slow email sending
    print(f"Email sent to {email}: Welcome, {name}!")


@app.post("/register")
def register(
    name:             str,
    email:            str,
    background_tasks: BackgroundTasks,
):
    # This returns immediately
    background_tasks.add_task(send_welcome_email, email, name)
    return {"message": f"Registered {name}. Welcome email on its way!"}

Deploying FastAPI

Option 1: Render

# Procfile or render.yaml start command:
uvicorn main:app --host 0.0.0.0 --port $PORT

Option 2: 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 build -t myapi .
docker run -p 8000:8000 myapi

Option 3: Multiple workers with Gunicorn

pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

Flask vs FastAPI — When to Use Which

Use Flask when:

  • Building a server-rendered website with HTML templates
  • You need a simple API with no special requirements
  • Your team already knows Flask
  • You need the massive Flask ecosystem (extensions)

Use FastAPI when:

  • Building a REST or GraphQL API
  • You want automatic validation and docs
  • Performance matters
  • You're using async Python
  • You're starting a new project today

What You Learned in This Chapter

  • FastAPI() creates the app. uvicorn main:app --reload runs it.
  • Type hints on function parameters do triple duty: URL parsing, validation, and docs generation.
  • response_model= controls what fields are returned — use it to hide sensitive data.
  • HTTPException(status_code=..., detail=...) returns clean error responses.
  • Pydantic BaseModel subclasses define request and response shapes. Field(...) adds validation rules. @field_validator adds custom checks.
  • Depends() injects shared logic (auth, DB, pagination) into endpoints.
  • async def routes handle I/O-bound work efficiently.
  • JWT auth: OAuth2PasswordBearer + python-jose + passlib.
  • lifespan context manager handles startup/shutdown.
  • BackgroundTasks runs work after the response is sent.
  • Auto-generated docs at /docs (Swagger) and /redoc (ReDoc).

What's Next?

Chapter 42 covers NumPy — the numerical computing library at the heart of Python's data science ecosystem. You'll learn how to work with arrays, perform vectorized math, and process numerical data orders of magnitude faster than plain Python lists.

© 2026 Abhilash Sahoo. Python: Zero to Hero.