Most of my channel is Java and Spring Boot. For this video I wanted a break, so I built something in Python that a complete beginner can finish in one sitting: a to-do app with a real REST API, a real database and a web page on top. It comes to three files and under 130 lines of Python.
This post is the written companion to the video. Every code block below is copied from the project, and I ran every request in it against that code before publishing. If you prefer to read, you can build the whole thing from this page. If you prefer to watch, the video is right here.
What you'll build
A to-do list where you can add a task, tick it off, rename it and delete it. The page talks to an API you write yourself, and the API saves everything in a database, so your tasks are still there after a restart. FastAPI also gives you a free interactive docs page at /docs where you can test every endpoint from the browser.
| File | Its one job |
|---|---|
index.html | The to-do page, styled with Bootstrap. Its buttons send requests to the API. |
main.py | The API: the endpoints, and the route that serves the page. |
database.py | The database connection, the tasks table, and the request and response models. |
tasks.db | Your data. SQLite creates this file the first time the app starts. |
The whole flow, for one click: the page sends a request to main.py, FastAPI checks the data and runs the matching function, that function uses database.py to read or save the task, and the result comes back as JSON so the page can redraw the list.
CRUD in one table
Almost every app does four things to its data: create, read, update and delete. On the web each one has its own HTTP method. The URL stays /tasks; the method tells the server what to do. This table is the plan for the whole app.
| Action | Method | URL | What it does | Success code |
|---|---|---|---|---|
| Create | POST | /tasks | Adds a new task | 201 |
| Read | GET | /tasks | Lists tasks, 50 at a time by default | 200 |
| Read | GET | /tasks/{id} | Returns one task | 200 |
| Update | PATCH | /tasks/{id} | Changes only the fields you send | 200 |
| Delete | DELETE | /tasks/{id} | Removes a task | 204 |
I use PATCH rather than PUT for updates. With PATCH you send only what changed, for example {"completed": true}, and everything else stays as it was.
Step 1: Set up the project
You need Python 3.10 or newer. Make a folder called fastapi-crud, open a terminal in it, and create a virtual environment so this project's packages stay separate from everything else:
python3 -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate
On a Mac the command is python3, not python. Once the environment is active, your prompt shows (.venv). Now create requirements.txt with these two lines:
fastapi[standard]==0.162.0 sqlalchemy==2.0.55
fastapi[standard] brings in FastAPI plus the server and the fastapi command we use to run it. SQLAlchemy is the library our code uses to talk to the database. Install both:
pip install -r requirements.txt
Step 2: database.py — the connection
Two names that beginners mix up: SQLite is the database. It keeps everything in one file and comes built into Python, so there is nothing to install. SQLAlchemy is the library our code uses to talk to it. Here is the top of database.py:
import os from datetime import datetime, timezone from pydantic import BaseModel, ConfigDict, Field, field_validator from sqlalchemy import DateTime, String, create_engine from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker # A local SQLite file by default. In production, set DATABASE_URL to PostgreSQL. DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./tasks.db") is_sqlite = DATABASE_URL.startswith("sqlite") engine = create_engine(DATABASE_URL, pool_pre_ping=True, connect_args={"check_same_thread": False} if is_sqlite else {}) SessionLocal = sessionmaker(engine, expire_on_commit=False) def get_db(): """Gives each request its own database session and closes it afterwards.""" with SessionLocal() as db: yield db
DATABASE_URL says where the data lives. By default it is a file called tasks.db in the project folder; in production you point it at PostgreSQL instead and the rest of the code stays the same. The engine is SQLAlchemy's connection to that database. A session is one conversation with the database, and get_db() opens one for each request and closes it when the request ends. The with block does the closing for us.
Step 3: database.py — one table, three API models
The rest of the file defines two kinds of class with two different jobs:
# The "tasks" table in the database class Base(DeclarativeBase): pass class TaskRow(Base): __tablename__ = "tasks" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(200)) completed: Mapped[bool] = mapped_column(default=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) # What the API accepts and returns class TaskCreate(BaseModel): title: str = Field(min_length=1, max_length=200) completed: bool = False class TaskUpdate(BaseModel): # every field optional: send only what changed title: str | None = Field(None, min_length=1, max_length=200) completed: bool | None = None class Task(TaskCreate): model_config = ConfigDict(from_attributes=True) # can be built from a TaskRow id: int created_at: datetime @field_validator("created_at") @classmethod def as_utc(cls, value: datetime) -> datetime: # SQLite forgets the time zone return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
| Class | Kind | Job |
|---|---|---|
TaskRow | SQLAlchemy table | One row in the tasks table. Each attribute is a column; the database fills in the id. |
TaskCreate | Pydantic model | The JSON body for adding a task. The title can't be empty or longer than 200 characters. |
TaskUpdate | Pydantic model | The body for PATCH. Every field is optional, so you send only what changed. |
Task | Pydantic model | The response. It adds the id and created_at, which the user never chooses. |
FastAPI checks every request against these models for you. If the title is missing or empty, the request stops with a 422 and your function never runs. from_attributes=True lets FastAPI build a Task response straight from a TaskRow object, so the endpoints can just return the database row.
created_at comes back from SQLite without its time zone, so the API would return the same moment in two different formats depending on whether the task was just created or read back later. The validator marks every timestamp as UTC so the response is always consistent.Step 4: main.py — the API
Start with the setup and one helper:
import logging import os from pathlib import Path from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, Query from fastapi.responses import FileResponse from sqlalchemy import select, text from sqlalchemy.orm import Session from database import Base, Task, TaskCreate, TaskRow, TaskUpdate, engine, get_db logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("tasks") Base.metadata.create_all(engine) # creates the tasks table if it doesn't exist yet app = FastAPI(title="Task API") DB = Annotated[Session, Depends(get_db)] def find_task(db: Session, task_id: int) -> TaskRow: task = db.get(TaskRow, task_id) if task is None: raise HTTPException(404, f"Task {task_id} not found") return task
create_all creates the tasks table the first time the app starts. DB is a shortcut that means "give me a database session": any endpoint that lists db: DB as a parameter gets one handed in by FastAPI's Depends. If you come from Spring, it is the same idea as @Autowired. find_task looks a task up and raises a 404 if it doesn't exist, so get, update and delete all share one place for that error.
Anatomy of an endpoint
Every endpoint is built the same way, so learn one and you can read them all. Here is update:
@app.patch("/tasks/{task_id}", response_model=Task) def update_task(task_id: int, data: TaskUpdate, db: DB): task = find_task(db, task_id) for field, value in data.model_dump(exclude_unset=True, exclude_none=True).items(): setattr(task, field, value) db.commit() log.info("Updated task %s", task_id) return task
The decorator links PATCH on that URL to the function below it. FastAPI fills the parameters for you: task_id comes from the URL and is converted to an int, data comes from the JSON body and is checked against TaskUpdate, and db is the session. exclude_unset=True is what makes PATCH work: only the fields the client actually sent are copied onto the task. Then db.commit() saves it.
exclude_none=True, sending {"title": null} is quietly ignored instead of rejected. You still get a 200, and the title doesn't change. That is fine for this app, because a task always needs a title, but it's worth knowing if you later add a field that people should be able to clear.The rest of the endpoints
@app.get("/", include_in_schema=False) def home(): return FileResponse(Path(__file__).parent / "index.html") @app.get("/health") def health(db: DB): db.execute(text("SELECT 1")) # fails if the database can't be reached return {"status": "ok"} @app.get("/tasks", response_model=list[Task]) def list_tasks(db: DB, skip: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=100)): return db.scalars(select(TaskRow).order_by(TaskRow.id).offset(skip).limit(limit)).all() @app.get("/tasks/{task_id}", response_model=Task) def get_task(task_id: int, db: DB): return find_task(db, task_id) @app.post("/tasks", response_model=Task, status_code=201) def create_task(data: TaskCreate, db: DB): task = TaskRow(**data.model_dump()) db.add(task) db.commit() log.info("Created task %s", task.id) return task @app.delete("/tasks/{task_id}", status_code=204) def delete_task(task_id: int, db: DB): db.delete(find_task(db, task_id)) db.commit() log.info("Deleted task %s", task_id)
list_tasks takes skip and limit so a long list comes back in pages: 50 by default, 100 at most. create_task returns 201 instead of the default 200, and delete_task returns 204, which means "done, nothing to send back". /health is what a hosting platform calls to check the app is alive, and it also checks the database. home serves the web page, and include_in_schema=False keeps it out of the docs page. Every create, update and delete also writes a log line, so you can watch them happen in the terminal.
Step 5: index.html — the to-do page
The page is plain HTML with Bootstrap for the styling, loaded from a CDN, so it needs an internet connection. The interesting part is the script. One small function handles every call to the API:
// Calls the API and returns the JSON response, or throws the API's error message async function api(method, path, body) { const res = await fetch(path, { method, headers: { "Content-Type": "application/json" }, body: body && JSON.stringify(body), }); if (!res.ok) { const { detail } = await res.json().catch(() => ({})); // a server crash (500) replies with plain text, not JSON throw new Error(typeof detail === "string" ? detail : res.status === 422 ? "Please check what you typed." : `Server error (${res.status}). Check the terminal for details.`); } return res.status === 204 ? null : res.json(); } // Runs an action, then reloads the list so the page always matches the database async function run(action) { try { await action(); const tasks = await api("GET", "/tasks?limit=100"); list.replaceChildren(...tasks.map(taskRow)); const left = tasks.filter((t) => !t.completed).length; document.getElementById("summary").textContent = tasks.length ? `${left} of ${tasks.length} left to do` : "No tasks yet. Add one above."; errorBox.classList.add("d-none"); } catch (err) { errorBox.textContent = err instanceof TypeError ? "Can't reach the server. Check that it's running." : err.message; errorBox.classList.remove("d-none"); } }
api() sends the request and turns any error into a readable message. run() does an action and then reloads the whole list from the server, so the page always matches the database instead of trying to keep its own copy in sync. Each button is one line on top of that:
function taskRow(task) { const row = document.getElementById("row").content.firstElementChild.cloneNode(true); const [check, title, editButton, deleteButton] = row.children; title.textContent = task.title; // textContent, not innerHTML: titles are user input title.classList.toggle("text-decoration-line-through", task.completed); title.classList.toggle("text-secondary", task.completed); check.checked = task.completed; check.setAttribute("aria-label", `Done: ${task.title}`); check.onchange = () => run(() => api("PATCH", `/tasks/${task.id}`, { completed: check.checked })); editButton.onclick = () => { const newTitle = prompt("Edit task", task.title)?.trim(); if (newTitle && newTitle !== task.title) run(() => api("PATCH", `/tasks/${task.id}`, { title: newTitle })); }; deleteButton.onclick = () => run(() => api("DELETE", `/tasks/${task.id}`)); return row; }
Notice title.textContent rather than innerHTML. Task titles are typed by users, and textContent shows them as plain text, so a title containing HTML can't run as code on the page.
Step 6: Run it and try every endpoint
fastapi dev
fastapi dev finds main.py on its own and restarts the server every time you save a file. Open http://localhost:8000 for the to-do app and http://localhost:8000/docs for the API docs. Stop the server with Ctrl+C. For production you would use fastapi run instead.
In /docs, click an endpoint, then Try it out, then Execute. Go through them in this order and you will see all five status codes this app uses:
| Do this | You get | Meaning |
|---|---|---|
POST /tasks with {"title": "Learn FastAPI"} | 201 Created | A new task was saved. Note its id. |
GET /tasks | 200 OK | It worked, here's your data. |
PATCH /tasks/{id} with {"completed": true} | 200 OK | Only completed changed; the title is the same. |
DELETE /tasks/{id} | 204 No Content | Deleted, nothing to send back. |
GET /tasks/{id} again | 404 Not Found | That task is gone. |
POST /tasks with {} | 422 | The body breaks a rule: title is required. |
If you prefer the terminal, the same steps with curl (replace 1 with the id the first command returns):
curl -X POST localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Learn FastAPI"}' curl localhost:8000/tasks curl localhost:8000/tasks/1 curl -X PATCH localhost:8000/tasks/1 -H "Content-Type: application/json" -d '{"completed": true}' curl -X DELETE localhost:8000/tasks/1 -i
Moving to PostgreSQL
SQLite is perfect for learning, but for production use PostgreSQL. You don't change any code: install the driver, create an empty database, and set DATABASE_URL before starting the app. The tasks table is created automatically on startup.
pip install "psycopg[binary]" export DATABASE_URL="postgresql+psycopg://user:password@localhost:5432/tasks" fastapi run
Try this yourself: add a due date
Give every task an optional due date. It takes three small changes in database.py: one column in TaskRow, and one field in both TaskCreate and TaskUpdate. Task inherits from TaskCreate, so the response picks it up for free. Don't forget to import date from datetime.
tasks.db before you restart. create_all() only creates tables that don't exist yet; it never adds a column to a table that's already there. I tried it with the old file still in place, and creating a task failed with a 500 error because the due_date column didn't exist. After deleting tasks.db it worked first time. In a real project you would use a migration tool like Alembic for this, so you don't lose your data.Post your version in the comments on the video. I'll reply to the first few.
Source code and links
The full project, with a README covering setup, testing and PostgreSQL, is on GitHub: github.com/sprashantofficial/fastapi-crud. If something doesn't work for you, leave a comment on the video with the step you were on and the error you saw.
Watch the video on YouTube: FastAPI Tutorial for Beginners – Build a Complete CRUD App From Scratch.
Comments
Post a Comment