Lesson 5 - Guided Project: Persistent Tasks API

Welcome to the Persistent Tasks API#

This is the fifth version of the Task Manager you’ve been building all course, and it crosses a line the first four never could: its data lives on disk. In Modules 1 through 4 your tasks lived in a Python list, so every restart wiped them clean. Here you’ll assemble everything from this module — the model family (TaskBase, the Task table, TaskCreate, TaskUpdate), the engine, the session dependency, and full CRUD — into one coherent app backed by SQLite. Then you’ll prove the payoff: after creating tasks, you’ll open a brand-new engine and session on the same database file and watch the rows still sitting there, exactly as they would after a real server restart.

By the end of this project, you will be able to:

• Assemble a complete SQLModel app: models, engine, session dependency, and table creation
• Implement full CRUD with correct status codes, pagination, and 404 handling
• Run the whole flow end to end and read the real responses
• Prove that your data persists by reopening the database file independently

Everything below was built and run as one app. Let’s assemble it stage by stage.

Stage 1: Models, Engine, and the Session Dependency#

Start with the foundation. The model family is the same one you designed earlier in the module: TaskBase holds the shared fields, the Task table adds the primary key, TaskCreate is what clients send, and TaskUpdate makes every field optional for partial updates. Below them sit the engine, the session dependency, and the one call that creates the table.

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import SQLModel, Field, Session, create_engine, select

class TaskBase(SQLModel):
    title: str
    done: bool = False

class Task(TaskBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

class TaskCreate(TaskBase):
    pass

class TaskUpdate(SQLModel):
    title: str | None = None
    done: bool | None = None

engine = create_engine(
    "sqlite:///tasks.db",
    connect_args={"check_same_thread": False},
)

def get_session():
    with Session(engine) as session:
        yield session

SQLModel.metadata.create_all(engine)

app = FastAPI()

A few choices are worth calling out. The engine points at sqlite:///tasks.db, a single file that becomes your database; the check_same_thread option is the SQLite-specific setting from Lesson 1 that lets FastAPI hand a session to whichever thread serves a request. get_session is the yield dependency: it opens a session, lends it to an endpoint, and closes it afterward — even on errors. And rather than wiring up a startup event, we call SQLModel.metadata.create_all(engine) once at module load, exactly as in the earlier lessons. It’s simpler, it runs no deprecated event hooks, and it’s safe to run every time because it only creates tables that don’t already exist.

Stage 2: The Full CRUD Endpoints#

Now the five operations. Each one declares session: Session = Depends(get_session), so it’s handed a working database session with no setup of its own. Read them top to bottom — create, list, read one, update, delete — and notice how short each is once the foundation is in place.

@app.post("/tasks", response_model=Task, status_code=201)
def create_task(task: TaskCreate, session: Session = Depends(get_session)):
    db_task = Task.model_validate(task)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task

@app.get("/tasks", response_model=list[Task])
def list_tasks(offset: int = 0, limit: int = 20, session: Session = Depends(get_session)):
    tasks = session.exec(select(Task).offset(offset).limit(limit)).all()
    return tasks

@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 not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.patch("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task_update: TaskUpdate, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(task, key, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()

Each endpoint is a small, focused unit of work:

• create_task accepts a TaskCreate, turns it into a real Task row with Task.model_validate, then add / commit / refresh saves it and reads back the database-assigned id. It returns 201 Created.
• list_tasks runs select(Task) with offset and limit query parameters so the list is paginated — the default page is the first twenty rows.
• get_task uses session.get(Task, task_id), the fast primary-key lookup, and raises 404 when nothing comes back.
• update_task does a partial update: model_dump(exclude_unset=True) keeps only the fields the client actually sent, and setattr applies just those, leaving everything else untouched.
• delete_task removes the row and returns 204 No Content — success with an empty body.

That’s the whole API. Next we run it.

Stage 3: Run the Whole Flow#

To exercise every endpoint without a browser, we drive the app with FastAPI’s TestClient, which sends real requests through the app in-process. We start from a fresh database, create three tasks, list them, mark two done, delete one, and finally request a task that doesn’t exist:

from fastapi.testclient import TestClient
client = TestClient(app)

# Create three tasks
for t in [{"title": "Write the report"},
          {"title": "Email the team"},
          {"title": "Book the venue"}]:
    r = client.post("/tasks", json=t)
    print(r.status_code, r.json())

# List them
print(client.get("/tasks").json())

# Mark the first two done (partial update)
print(client.patch("/tasks/1", json={"done": True}).json())
print(client.patch("/tasks/2", json={"done": True}).json())

# Delete the third
r = client.delete("/tasks/3")
print(r.status_code)

# List again, then request a missing task
print(client.get("/tasks").json())
r = client.get("/tasks/99")
print(r.status_code, r.json())

Running it produces exactly the behavior the endpoints promise — fresh ids 1, 2, 3 on creation, a working partial update, a 204 on delete, and a clean 404 for the missing task:

=== CREATE 3 TASKS ===
201 {'done': False, 'id': 1, 'title': 'Write the report'}
201 {'done': False, 'id': 2, 'title': 'Email the team'}
201 {'done': False, 'id': 3, 'title': 'Book the venue'}

=== GET /tasks ===
200 [{'done': False, 'id': 1, 'title': 'Write the report'}, {'done': False, 'id': 2, 'title': 'Email the team'}, {'done': False, 'id': 3, 'title': 'Book the venue'}]

=== PATCH /tasks/1 {done: true} ===
200 {'done': True, 'id': 1, 'title': 'Write the report'}

=== PATCH /tasks/2 {done: true} ===
200 {'done': True, 'id': 2, 'title': 'Email the team'}

=== DELETE /tasks/3 ===
204 (no body)

=== GET /tasks (after delete) ===
200 [{'done': True, 'id': 1, 'title': 'Write the report'}, {'done': True, 'id': 2, 'title': 'Email the team'}]

=== GET /tasks/99 (missing) ===
404 {'detail': 'Task not found'}

Look closely at the PATCH results: setting done to True left each title exactly as it was. That’s exclude_unset doing its job — only the field the client sent changed. After the delete, the list is [1, 2], and the missing task returns the structured {'detail': 'Task not found'} error FastAPI builds from HTTPException.

Stage 4: Prove the Data Persists#

Everything so far would have worked with an in-memory list too. The real difference is what happens to the data when the program is gone. Those rows now live inside tasks.db on disk. To prove it, we open a completely fresh engine on the same file — no shared state with the app above — start a new session, and query the table directly. This simulates exactly what a restarted server would see:

from sqlmodel import create_engine, Session, select

# A brand-new engine on the SAME file — nothing shared with the running app
fresh_engine = create_engine("sqlite:///tasks.db")
with Session(fresh_engine) as session:
    rows = session.exec(select(Task)).all()
    print([(t.id, t.title, t.done) for t in rows])

[(1, 'Write the report', True), (2, 'Email the team', True)]

There they are. Task 3 is gone because we deleted it, tasks 1 and 2 are still marked done, and all of it came back from a fresh connection that never touched the original app objects. That is persistence: the data outlives the process. Restart your machine, reopen the file tomorrow, and the same rows are waiting.

To run this as a real server instead of through TestClient, save the app in main.py and start it:

uvicorn main:app --reload

Then open http://127.0.0.1:8000/docs and try every endpoint from the interactive Swagger UI — POST a task, list them, PATCH one, DELETE another. Stop the server with Ctrl+C, start it again, and reload /docs: your tasks are still there, because tasks.db was on disk the whole time.

Practice Exercises#

Exercise 1: Add a created_at timestamp#

Give every task a created_at column that records when it was created, so you can sort or display the age of each task. Add the field to Task (not to TaskCreate — the server sets it, not the client) and confirm new rows come back with a timestamp.

Exercise 2: Filter the list by done#

Add an optional done query parameter to list_tasks so a client can request only finished or only unfinished tasks — for example GET /tasks?done=true.

Exercise 3: Add a count endpoint#

Add GET /tasks/count that returns the total number of tasks, e.g. {"count": 2}, without fetching every row’s data.

Summary#

You assembled the fifth Task Manager into one coherent app backed by a real database. The model family (TaskBase, Task, TaskCreate, TaskUpdate) gave you separate shapes for storage, input, and updates; the engine and the get_session dependency connected each request to SQLite; and five short endpoints delivered full CRUD with the right status codes (201, 200, 204), pagination, partial updates via exclude_unset, and 404s through HTTPException. Most importantly, you proved persistence: a fresh engine opened on the same file returned the surviving rows, demonstrating that your data outlives the process.

Key Concepts#

• Assembled app — models, engine, session dependency, and create_all form the foundation that CRUD sits on.
• Full CRUD — POST 201, paginated GET list, GET one with 404, PATCH partial update, DELETE 204.
• Partial update — model_dump(exclude_unset=True) plus setattr changes only the fields a client sent.
• Persistence — rows live in tasks.db on disk and survive restarts; a fresh engine reads them back.
• SQLite to PostgreSQL — swap the connection string in production; models and queries stay the same.

Why This Matters#

This is the moment your project stops being a demo and becomes something real. Data that disappears on restart is fine for a tutorial but useless for an application people depend on. The engine-plus-session-dependency pattern with full CRUD is the standard FastAPI shape for persistent APIs, and you’ll reuse it on nearly every backend you build. With storage solved, the next concern is who is allowed to touch it.

Next Steps#

Continue Building Your Skills#

You now have a Task Manager whose data survives restarts — a complete, persistent CRUD API built entirely from this module’s pieces. The natural next step is control: right now anyone who reaches the API can create, edit, and delete tasks. In the next module you’ll add authentication and security, putting these endpoints behind a login so your persistent data is also protected data.