FastAPI is the quickest way to build a proper REST API in Python: you declare what a request should look like, and it validates the input, produces the documentation and returns clear errors for you. This project builds a small task API from an empty folder to a tested, database-backed service in five stages. Every stage runs; nothing is left as an exercise.
Stage 1: hello#
mkdir task-api && cd task-api
python -m venv .venv && source .venv/bin/activate # .venv\Scripts\activate on Windows
pip install "fastapi[standard]" sqlmodel pytest
# main.py
from fastapi import FastAPI
app = FastAPI(title="Task API")
@app.get("/")
def root():
return {"status": "ok"}
fastapi dev main.py
Open http://127.0.0.1:8000 for the JSON, and http://127.0.0.1:8000/docs for interactive documentation that FastAPI generated from the code. That docs page is where you will test every endpoint from here on.
Stage 2: a model and validation#
from pydantic import BaseModel, Field
class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
done: bool = False
class Task(TaskCreate):
id: int
A request body declared as TaskCreate is parsed and validated before your function runs. Send a missing title, or a title of 300 characters, and the client gets a 422 with a message naming the field. You wrote no validation code.
Stage 3: CRUD in memory#
from fastapi import FastAPI, HTTPException, status
app = FastAPI(title="Task API")
tasks: dict[int, Task] = {}
next_id = 1
@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate):
global next_id
task = Task(id=next_id, **payload.model_dump())
tasks[task.id] = task
next_id += 1
return task
@app.get("/tasks", response_model=list[Task])
def list_tasks(done: bool | None = None):
items = list(tasks.values())
if done is not None:
items = [t for t in items if t.done == done]
return items
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int):
if task_id not in tasks:
raise HTTPException(status_code=404, detail="Task not found")
return tasks[task_id]
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, payload: TaskCreate):
if task_id not in tasks:
raise HTTPException(status_code=404, detail="Task not found")
tasks[task_id] = Task(id=task_id, **payload.model_dump())
return tasks[task_id]
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int):
if task_id not in tasks:
raise HTTPException(status_code=404, detail="Task not found")
del tasks[task_id]
Notice the details that make it a real API rather than a demo: 201 on create, 204 with no body on delete, 404 with a message when the id does not exist, and a query parameter ?done=true that FastAPI converts and validates. The docs page now shows all five routes with try-it-out buttons.
Stage 4: a database#
The in-memory dictionary forgets everything on restart. SQLModel, from the same author as FastAPI, combines Pydantic models with SQLAlchemy tables so one class does both jobs:
# models.py
from sqlmodel import SQLModel, Field
class TaskBase(SQLModel):
title: str = Field(min_length=1, max_length=200)
done: bool = False
class Task(TaskBase, table=True):
id: int | None = Field(default=None, primary_key=True)
class TaskCreate(TaskBase):
pass
# db.py
from sqlmodel import SQLModel, Session, create_engine
engine = create_engine("sqlite:///tasks.db")
def init_db():
SQLModel.metadata.create_all(engine)
def get_session():
with Session(engine) as session:
yield session
# main.py
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException, status
from sqlmodel import Session, select
from db import init_db, get_session
from models import Task, TaskCreate
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
yield
app = FastAPI(title="Task API", lifespan=lifespan)
@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: Session = Depends(get_session)):
task = Task.model_validate(payload)
session.add(task)
session.commit()
session.refresh(task)
return task
@app.get("/tasks", response_model=list[Task])
def list_tasks(done: bool | None = None, session: Session = Depends(get_session)):
query = select(Task)
if done is not None:
query = query.where(Task.done == done)
return session.exec(query).all()
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
return task
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, payload: TaskCreate, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
task.title, task.done = payload.title, payload.done
session.add(task)
session.commit()
session.refresh(task)
return task
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, session: Session = Depends(get_session)):
task = session.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
session.delete(task)
session.commit()
Depends(get_session) is FastAPI’s dependency injection: each request gets a database session that is opened before the handler and closed after. Swapping SQLite for PostgreSQL is one change to the connection string.
Stage 5: tests#
# test_api.py
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, Session, create_engine
from sqlmodel.pool import StaticPool
import pytest
from main import app
from db import get_session
@pytest.fixture
def client():
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
SQLModel.metadata.create_all(engine)
def override():
with Session(engine) as s:
yield s
app.dependency_overrides[get_session] = override
yield TestClient(app)
app.dependency_overrides.clear()
def test_create_and_fetch(client):
created = client.post("/tasks", json={"title": "Write tests"})
assert created.status_code == 201
task_id = created.json()["id"]
fetched = client.get("/tasks/" + str(task_id))
assert fetched.json()["title"] == "Write tests"
def test_validation(client):
assert client.post("/tasks", json={"title": ""}).status_code == 422
def test_missing(client):
assert client.get("/tasks/999").status_code == 404
pytest -q
The tests use an in-memory database by overriding the session dependency, so they never touch tasks.db and run in a fraction of a second.
Where to take it#
- Authentication with a bearer token, using FastAPI’s security utilities.
- Pagination on the list endpoint:
limitandoffsetquery parameters. - A
PATCHroute that updates only the fields sent. - Deploy it: a Dockerfile with
fastapi run, and PostgreSQL instead of SQLite.
Questions people ask#
Why FastAPI rather than Flask?
Validation, documentation and async support come built in. Flask is smaller and does less by default; for an API, FastAPI’s defaults are what you would end up adding anyway.
Is SQLModel required?
No. Plain SQLAlchemy, or any database library, works with the same dependency pattern. SQLModel simply avoids defining each model twice.
What does fastapi dev do?
Runs the app with auto-reload for development. fastapi run is the production form without reload.
Where does the docs page come from?
FastAPI builds an OpenAPI description from your routes and models and serves it with a Swagger UI at /docs and ReDoc at /redoc. It updates every time the code changes.
Where to go next#
- FastAPI vs Flask — the decision this project assumes.
- REST API error handling — the status codes used here, in depth.
- A CRUD app with Flask and SQLAlchemy — the same idea in the other framework.
