Chapter 32: Web Development with Flask and FastAPI
Every web application — from a simple API to a complex platform — boils down to the same loop: receive a request, do some work, send a response. Python has excellent tools for this. Two dominate the ecosystem:
- Flask — lightweight, explicit, minimal magic. You control everything. Great for learning and small-to-medium apps.
- FastAPI — modern, fast, built on type hints and async. Generates API documentation automatically. The choice for new production APIs in 2026.
This chapter builds real applications with both. By the end you'll have a working REST API backed by a database with authentication.
HTTP in 60 Seconds
Every web request has:
- A method — what you want to do (
GET,POST,PUT,PATCH,DELETE) - A URL — where to do it (
/users/42) - Headers — metadata (
Content-Type: application/json,Authorization: Bearer token) - A body — data sent with the request (only for
POST,PUT,PATCH)
Every response has:
- A status code — what happened (
200 OK,201 Created,400 Bad Request,404 Not Found,500 Internal Server Error) - Headers — metadata about the response
- A body — the data (usually JSON for APIs)
REST API conventions:
| Method | URL | Action |
|---|---|---|
GET |
/users |
List all users |
GET |
/users/42 |
Get user 42 |
POST |
/users |
Create a new user |
PUT |
/users/42 |
Replace user 42 entirely |
PATCH |
/users/42 |
Update user 42 partially |
DELETE |
/users/42 |
Delete user 42 |
Flask
pip install flask
Hello, Flask
# app.py
from flask import Flask
app = Flask(__name__)
@app.route("/")
def home():
return "Hello, World!"
@app.route("/hello/<name>")
def hello(name):
return f"Hello, {name}!"
if __name__ == "__main__":
app.run(debug=True)
python app.py
# * Running on http://127.0.0.1:5000
Visit http://localhost:5000/hello/Alice -> Hello, Alice!
@app.route registers a URL pattern. <name> is a variable part of the URL, passed as a parameter to the function.
Returning JSON
from flask import Flask, jsonify, request, abort
app = Flask(__name__)
# In-memory store for this example
users = {
1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
2: {"id": 2, "name": "Bob", "email": "bob@example.com"},
}
next_id = 3
@app.route("/users", methods=["GET"])
def list_users():
return jsonify(list(users.values()))
@app.route("/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
user = users.get(user_id)
if user is None:
abort(404) # raises 404 Not Found
return jsonify(user)
@app.route("/users", methods=["POST"])
def create_user():
global next_id
data = request.get_json()
if not data or "name" not in data or "email" not in data:
abort(400) # Bad Request
user = {"id": next_id, "name": data["name"], "email": data["email"]}
users[next_id] = user
next_id += 1
return jsonify(user), 201 # 201 Created
@app.route("/users/<int:user_id>", methods=["PUT"])
def update_user(user_id):
if user_id not in users:
abort(404)
data = request.get_json()
if not data:
abort(400)
users[user_id].update(data)
return jsonify(users[user_id])
@app.route("/users/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
if user_id not in users:
abort(404)
del users[user_id]
return "", 204 # 204 No Content
Error handlers
@app.errorhandler(404)
def not_found(e):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(400)
def bad_request(e):
return jsonify({"error": "Bad request"}), 400
@app.errorhandler(500)
def server_error(e):
return jsonify({"error": "Internal server error"}), 500
Request object
from flask import request
@app.route("/search")
def search():
# Query parameters: /search?q=python&limit=10
query = request.args.get("q", "")
limit = request.args.get("limit", 10, type=int)
# Headers
auth = request.headers.get("Authorization", "")
ct = request.headers.get("Content-Type", "")
# JSON body
data = request.get_json()
# Form data
field = request.form.get("username")
return jsonify({"query": query, "limit": limit})
Blueprints — organizing a larger app
# users/routes.py
from flask import Blueprint, jsonify
users_bp = Blueprint("users", __name__, url_prefix="/users")
@users_bp.route("/")
def list_users():
return jsonify([])
@users_bp.route("/<int:user_id>")
def get_user(user_id):
return jsonify({"id": user_id})
# orders/routes.py
from flask import Blueprint
orders_bp = Blueprint("orders", __name__, url_prefix="/orders")
@orders_bp.route("/")
def list_orders():
return jsonify([])
# app.py
from flask import Flask
from users.routes import users_bp
from orders.routes import orders_bp
app = Flask(__name__)
app.register_blueprint(users_bp)
app.register_blueprint(orders_bp)
A complete Flask API with SQLite
"""
flask_api.py — A REST API for a task manager, backed by SQLite.
"""
import sqlite3
from flask import Flask, jsonify, request, abort, g
from pathlib import Path
app = Flask(__name__)
DB_PATH = Path("tasks.db")
# ── Database helpers ──────────────────────────────────────────────────────────
def get_db():
"""Return a database connection, reusing the one for this request."""
if "db" not in g:
g.db = sqlite3.connect(DB_PATH)
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA journal_mode=WAL")
return g.db
@app.teardown_appcontext
def close_db(error):
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
with app.app_context():
db = get_db()
db.executescript("""
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT DEFAULT '',
done INTEGER DEFAULT 0,
created_at TEXT DEFAULT (datetime('now'))
);
""")
db.commit()
# ── Routes ─────────────────────────────────────────────────────────────────────
@app.route("/tasks", methods=["GET"])
def list_tasks():
done = request.args.get("done")
query = "SELECT * FROM tasks"
if done is not None:
query += f" WHERE done = {1 if done == 'true' else 0}"
query += " ORDER BY created_at DESC"
tasks = get_db().execute(query).fetchall()
return jsonify([dict(t) for t in tasks])
@app.route("/tasks/<int:task_id>", methods=["GET"])
def get_task(task_id):
task = get_db().execute(
"SELECT * FROM tasks WHERE id = ?", (task_id,)
).fetchone()
if task is None:
abort(404)
return jsonify(dict(task))
@app.route("/tasks", methods=["POST"])
def create_task():
data = request.get_json(silent=True) or {}
title = data.get("title", "").strip()
if not title:
return jsonify({"error": "title is required"}), 400
db = get_db()
cursor = db.execute(
"INSERT INTO tasks (title, description) VALUES (?, ?)",
(title, data.get("description", ""))
)
db.commit()
task = db.execute(
"SELECT * FROM tasks WHERE id = ?", (cursor.lastrowid,)
).fetchone()
return jsonify(dict(task)), 201
@app.route("/tasks/<int:task_id>", methods=["PATCH"])
def update_task(task_id):
task = get_db().execute(
"SELECT id FROM tasks WHERE id = ?", (task_id,)
).fetchone()
if task is None:
abort(404)
data = request.get_json(silent=True) or {}
allowed = {"title", "description", "done"}
updates = {k: v for k, v in data.items() if k in allowed}
if not updates:
return jsonify({"error": "No valid fields to update"}), 400
set_clause = ", ".join(f"{k} = ?" for k in updates)
db = get_db()
db.execute(
f"UPDATE tasks SET {set_clause} WHERE id = ?",
[*updates.values(), task_id]
)
db.commit()
updated = db.execute(
"SELECT * FROM tasks WHERE id = ?", (task_id,)
).fetchone()
return jsonify(dict(updated))
@app.route("/tasks/<int:task_id>", methods=["DELETE"])
def delete_task(task_id):
db = get_db()
cursor = db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
db.commit()
if cursor.rowcount == 0:
abort(404)
return "", 204
@app.errorhandler(404)
def not_found(e):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(400)
def bad_request(e):
return jsonify({"error": "Bad request"}), 400
if __name__ == "__main__":
init_db()
app.run(debug=True)
Test it with curl:
# Create
curl -X POST http://localhost:5000/tasks \
-H "Content-Type: application/json" \
-d '{"title": "Learn FastAPI", "description": "Read the docs"}'
# List
curl http://localhost:5000/tasks
# Get one
curl http://localhost:5000/tasks/1
# Update
curl -X PATCH http://localhost:5000/tasks/1 \
-H "Content-Type: application/json" \
-d '{"done": 1}'
# Delete
curl -X DELETE http://localhost:5000/tasks/1
FastAPI
FastAPI is built on type hints, Pydantic models, and asyncio. It auto-generates interactive API documentation, validates request and response data automatically, and is one of the fastest Python frameworks available.
pip install fastapi uvicorn[standard]
Hello, FastAPI
# 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("/hello/{name}")
def hello(name: str):
return {"message": f"Hello, {name}!"}
uvicorn main:app --reload
# Open http://localhost:8000/docs <- interactive Swagger UI, auto-generated
# Open http://localhost:8000/redoc <- ReDoc documentation
Pydantic models for validation
FastAPI uses Pydantic for data validation. Define a model and FastAPI validates incoming JSON automatically:
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr, Field, field_validator
from typing import Optional
from datetime import datetime
app = FastAPI()
class UserCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
email: str = Field(..., pattern=r"[^@]+@[^@]+\.[^@]+")
age: int = Field(..., ge=0, le=150)
bio: Optional[str] = None
@field_validator("name")
@classmethod
def name_must_not_be_blank(cls, v: str) -> str:
if not v.strip():
raise ValueError("Name cannot be blank.")
return v.strip()
class UserResponse(BaseModel):
id: int
name: str
email: str
age: int
created_at: datetime
model_config = {"from_attributes": True} # allow ORM objects
# In-memory store
users: dict[int, dict] = {}
next_id = 1
@app.get("/users", response_model=list[UserResponse])
def list_users():
return list(users.values())
@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
if user_id not in users:
raise HTTPException(status_code=404, detail="User not found")
return users[user_id]
@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
global next_id
new_user = {
"id": next_id,
"name": user.name,
"email": user.email,
"age": user.age,
"created_at": datetime.utcnow(),
}
users[next_id] = new_user
next_id += 1
return new_user
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
if user_id not in users:
raise HTTPException(status_code=404, detail="User not found")
del users[user_id]
If you POST invalid data — a negative age, a malformed email — FastAPI returns a detailed 422 Unprocessable Entity response automatically, with the exact field and error message. No validation code needed in your route handlers.
Query parameters and filters
from fastapi import Query
@app.get("/users")
def list_users(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
active: Optional[bool] = None,
search: Optional[str] = None,
):
result = list(users.values())
if active is not None:
result = [u for u in result if u.get("active") == active]
if search:
result = [u for u in result if search.lower() in u["name"].lower()]
return result[skip : skip + limit]
Path parameters with validation
from fastapi import Path
@app.get("/users/{user_id}")
def get_user(user_id: int = Path(..., ge=1)):
...
@app.get("/items/{item_name}")
def get_item(item_name: str = Path(..., min_length=1, max_length=50)):
...
Dependency injection
FastAPI's dependency injection system is one of its most powerful features. Dependencies are functions that run before your route handler and provide resources:
from fastapi import Depends
import sqlite3
def get_db():
conn = sqlite3.connect("app.db")
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
@app.get("/tasks")
def list_tasks(db: sqlite3.Connection = Depends(get_db)):
return db.execute("SELECT * FROM tasks").fetchall()
@app.post("/tasks")
def create_task(task: TaskCreate, db: sqlite3.Connection = Depends(get_db)):
cursor = db.execute(
"INSERT INTO tasks (title) VALUES (?)", (task.title,)
)
db.commit()
return {"id": cursor.lastrowid, "title": task.title}
Async routes
FastAPI supports both sync and async route handlers. Use async def when calling async I/O (databases, HTTP, file I/O):
import asyncio
import httpx
@app.get("/weather/{city}")
async def get_weather(city: str):
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.open-meteo.com/v1/forecast",
params={"latitude": 51.5, "longitude": -0.1, "current_weather": True}
)
return response.json()
A complete FastAPI app with SQLAlchemy
"""
fastapi_app.py — A production-ready task API with FastAPI + SQLAlchemy.
"""
from __future__ import annotations
from datetime import datetime
from typing import Optional
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text, create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
# ── Database ──────────────────────────────────────────────────────────────────
class Base(DeclarativeBase):
pass
class TaskModel(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(200), nullable=False)
description = Column(Text, default="")
done = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
engine = create_engine("sqlite:///tasks.db", connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ── Pydantic schemas ──────────────────────────────────────────────────────────
class TaskCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
description: str = Field("", max_length=2000)
class TaskUpdate(BaseModel):
title: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=2000)
done: Optional[bool] = None
class TaskOut(BaseModel):
id: int
title: str
description: str
done: bool
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
# ── App ───────────────────────────────────────────────────────────────────────
app = FastAPI(
title = "Task Manager API",
description = "A production-ready task manager.",
version = "1.0.0",
)
@app.get("/tasks", response_model=list[TaskOut], tags=["Tasks"])
def list_tasks(
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
done: Optional[bool]= Query(None),
search: Optional[str] = Query(None, min_length=1),
db: Session = Depends(get_db),
):
"""List tasks with optional filtering and pagination."""
q = db.query(TaskModel)
if done is not None:
q = q.filter(TaskModel.done == done)
if search:
q = q.filter(TaskModel.title.ilike(f"%{search}%"))
return q.order_by(TaskModel.created_at.desc()).offset(skip).limit(limit).all()
@app.get("/tasks/{task_id}", response_model=TaskOut, tags=["Tasks"])
def get_task(task_id: int, db: Session = Depends(get_db)):
"""Get a single task by ID."""
task = db.get(TaskModel, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@app.post("/tasks", response_model=TaskOut, status_code=201, tags=["Tasks"])
def create_task(payload: TaskCreate, db: Session = Depends(get_db)):
"""Create a new task."""
task = TaskModel(**payload.model_dump())
db.add(task)
db.commit()
db.refresh(task)
return task
@app.patch("/tasks/{task_id}", response_model=TaskOut, tags=["Tasks"])
def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_db)):
"""Partially update a task."""
task = db.get(TaskModel, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
for field, value in payload.model_dump(exclude_unset=True).items():
setattr(task, field, value)
task.updated_at = datetime.utcnow()
db.commit()
db.refresh(task)
return task
@app.delete("/tasks/{task_id}", status_code=204, tags=["Tasks"])
def delete_task(task_id: int, db: Session = Depends(get_db)):
"""Delete a task."""
task = db.get(TaskModel, task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
db.delete(task)
db.commit()
@app.get("/stats", tags=["Stats"])
def stats(db: Session = Depends(get_db)):
"""Return task statistics."""
total = db.query(TaskModel).count()
done = db.query(TaskModel).filter(TaskModel.done == True).count()
pending = total - done
return {"total": total, "done": done, "pending": pending}
if __name__ == "__main__":
uvicorn.run("fastapi_app:app", host="0.0.0.0", port=8000, reload=True)
Run it, then visit http://localhost:8000/docs for the auto-generated Swagger UI where you can test every endpoint interactively, right in the browser.
Adding Authentication
The most common pattern for API authentication is JWT (JSON Web Tokens):
pip install python-jose[cryptography] passlib[bcrypt]
from datetime import datetime, timedelta
from typing import Optional
from fastapi import 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 = "your-secret-key-store-in-env-not-in-code"
ALGORITHM = "HS256"
TOKEN_EXPIRY_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
payload = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
payload.update({"exp": expire})
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def decode_token(token: str) -> dict:
try:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
headers={"WWW-Authenticate": "Bearer"},
)
# Fake user database
fake_users = {
"alice": {
"username": "alice",
"hashed_password": hash_password("secret"),
"email": "alice@example.com",
}
}
@app.post("/auth/token", tags=["Auth"])
def login(form: OAuth2PasswordRequestForm = Depends()):
user = fake_users.get(form.username)
if not user or not verify_password(form.password, user["hashed_password"]):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
token = create_token({"sub": form.username},
timedelta(minutes=TOKEN_EXPIRY_MINUTES))
return {"access_token": token, "token_type": "bearer"}
def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
payload = decode_token(token)
username = payload.get("sub")
user = fake_users.get(username)
if not user:
raise HTTPException(status_code=401, detail="User not found")
return user
# Protect a route
@app.get("/me", tags=["Auth"])
def get_me(current_user: dict = Depends(get_current_user)):
return {"username": current_user["username"], "email": current_user["email"]}
@app.get("/tasks/mine", tags=["Tasks"])
def my_tasks(
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_db),
):
# Filter tasks by the current user
return db.query(TaskModel).filter(
TaskModel.owner == current_user["username"]
).all()
Flask vs FastAPI — When to Use Which
| Feature | Flask | FastAPI |
|---|---|---|
| Learning curve | Lower | Slightly higher |
| Auto-generated docs | No (use Flask-RESTX) | Yes — Swagger + ReDoc |
| Data validation | Manual | Automatic (Pydantic) |
| Type safety | Optional | First-class |
| Async support | Limited (Flask 2+) | Native |
| Performance | Good | Excellent (one of the fastest) |
| Ecosystem maturity | Very mature (since 2010) | Growing fast (2018+) |
| Best for | Prototypes, full-stack, legacy | New APIs, production REST |
Rule of thumb: If you're building a new REST API in 2026, use FastAPI. If you're maintaining an existing Flask app or building a full-stack app with Jinja2 templates, use Flask.
Project Structure for a Real API
my_api/
├── app/
│ ├── __init__.py
│ ├── main.py <- FastAPI app creation
│ ├── config.py <- Settings from environment variables
│ ├── database.py <- Engine, SessionLocal, get_db
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── task.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py <- Pydantic models for User
│ │ └── task.py <- Pydantic models for Task
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── users.py
│ │ ├── tasks.py
│ │ └── auth.py
│ ├── services/
│ │ ├── user_service.py
│ │ └── task_service.py
│ └── dependencies.py <- get_db, get_current_user
├── tests/
│ ├── conftest.py
│ ├── test_users.py
│ └── test_tasks.py
├── alembic/ <- Database migrations
├── requirements.txt
├── .env <- SECRET_KEY, DATABASE_URL (never commit this)
└── README.md
# app/config.py — settings from environment variables
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str = "sqlite:///./app.db"
secret_key: str = "change-me-in-production"
token_expire_min:int = 30
debug: bool = False
class Config:
env_file = ".env"
settings = Settings()
# app/routers/tasks.py — a proper router module
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import Optional
from ..database import get_db
from ..models.task import TaskModel
from ..schemas.task import TaskCreate, TaskUpdate, TaskOut
from ..dependencies import get_current_user
router = APIRouter(prefix="/tasks", tags=["Tasks"])
@router.get("/", response_model=list[TaskOut])
def list_tasks(
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: Session = Depends(get_db),
):
return db.query(TaskModel).offset(skip).limit(limit).all()
# ... other endpoints
# app/main.py
from fastapi import FastAPI
from .routers import tasks, users, auth
app = FastAPI(title="My API")
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(tasks.router)
What You Learned in This Chapter
- HTTP has methods (
GET,POST,PUT,PATCH,DELETE), status codes (200,201,400,404,500), and a URL + body structure. REST APIs follow a standard URL naming convention. - Flask registers routes with
@app.route.request.get_json()reads the body.jsonify()returns JSON.abort(404)raises HTTP errors. Blueprints organize routes into separate files. - FastAPI uses type hints and Pydantic models for automatic request validation and response serialization. Bad data returns a detailed
422error automatically. Field(...)sets validation rules —min_length,ge,le,pattern, etc.Query(...)andPath(...)validate query parameters and path parameters.Depends(get_db)injects a database session using FastAPI's dependency injection.async defroutes handle async I/O natively in FastAPI.- JWT tokens with
python-joseandpasslibimplement secure authentication.OAuth2PasswordBearerwires it into FastAPI's docs UI. - Run FastAPI with
uvicorn main:app --reload. Docs appear at/docsand/redoc. - Structure a real API into
models/,schemas/,routers/,services/, and load settings from.envwithpydantic-settings.
What's Next?
Chapter 33 covers Data Science with Python — numpy, pandas, matplotlib, and scikit-learn. Whether you're analyzing sales data, visualizing trends, or training a machine learning model, Python's data science stack is unmatched.