Lesson 3 - CRUD: Create and Read

Welcome to CRUD: Create and Read#

You now have a database connected (Lesson 1) and clean table and schema models to work with (Lesson 2). It’s time to actually move data. The four operations every database app needs are Create, Read, Update, and Delete — together, CRUD. This lesson covers the first two: writing new rows and reading them back. By the end, your Task Manager will persist tasks to disk and serve them on request, surviving any restart.

The good news is that with the foundation in place, each operation is just a few lines. You’ll add three rows to a real SQLite file, fetch one by its id, and page through the full list — all through the session dependency you already built.

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

• Create a row by converting a create schema with Task.model_validate and saving it with add, commit, and refresh
• Read one row by id with session.get, returning a 404 when it’s missing
• Read many rows with a select query, using offset and limit for pagination
• Wire all three as endpoints through the get_session dependency and a response_model

Let’s write some data.

Setting Up the App#

Every example below shares the same setup: the table and schema models from Lesson 2, the engine, and the session dependency from Lesson 1. Here is the full app we’ll build piece by piece, so you have the whole picture before we zoom in:

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

engine = create_engine("sqlite:///tasks.db", connect_args={"check_same_thread": False})
def get_session():
    with Session(engine) as session:
        yield session

app = FastAPI()
SQLModel.metadata.create_all(engine)

TaskCreate is what clients send (just title and done — no id); Task is the table that gains a database-assigned id. With this in place, the three endpoints follow.

Create: Writing a New Row#

To create a task, the endpoint receives a TaskCreate, turns it into a Task table row, and saves it. The conversion is one line — Task.model_validate(data) — and saving is the three-step add, commit, refresh sequence:

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

Task.model_validate(data) reads the fields off the incoming TaskCreate and builds a real Task instance — a row object the database understands. At this moment its id is still None, because the database hasn’t assigned one yet:

data = TaskCreate(title="Write the report")
task = Task.model_validate(data)
print("id before commit:", task.id)

id before commit: None

Then the three database calls do their distinct jobs, and afterward the id is filled in. POSTing three tasks gives us:

# POST /tasks  {"title": "Write the report"}
# POST /tasks  {"title": "Email the team", "done": true}
# POST /tasks  {"title": "Plan the sprint"}

POST 201 {'done': False, 'id': 1, 'title': 'Write the report'}
POST 201 {'done': True, 'id': 2, 'title': 'Email the team'}
POST 201 {'done': False, 'id': 3, 'title': 'Plan the sprint'}

Each response carries a real id (1, 2, 3) that the database generated. The status_code=201 marks “created,” and response_model=Task shapes the response. That id is the value clients will use to read, update, or delete the task later.

Read One: Fetch by Id, or 404#

Reading a single row by its primary key is the simplest query SQLModel offers: session.get(Task, task_id). It returns the matching Task, or None if no row has that id. A well-behaved API turns that None into a clear 404 Not Found rather than crashing:

@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

Fetching an existing task returns it directly; asking for one that doesn’t exist returns a 404 with a helpful message:

# GET /tasks/2
# GET /tasks/99

GET /tasks/2 200 {'done': True, 'id': 2, 'title': 'Email the team'}
GET /tasks/99 404 {'detail': 'Task not found'}

session.get is purpose-built for lookups by primary key, so it’s both clearer and faster than writing a full query for this case. The if not task guard is the standard pattern you’ll repeat in the update and delete endpoints next lesson — always check before you trust the row exists.

Read Many: A Paginated List#

To return all tasks, you build a query with select(Task) and run it through the session. But returning every row at once is a bad idea once a table grows large, so we add pagination: offset skips a number of rows and limit caps how many come back. We expose those as query parameters with sensible defaults:

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

session.exec(...) runs the query, and .all() collects the results into a list. With no parameters, you get the first page (up to 10 rows); the response_model=list[Task] tells FastAPI to serialize each row:

# GET /tasks

GET /tasks 200 [{'done': False, 'id': 1, 'title': 'Write the report'}, {'done': True, 'id': 2, 'title': 'Email the team'}, {'done': False, 'id': 3, 'title': 'Plan the sprint'}]

Add skip and limit to walk through pages. Asking to skip the first row and take just one returns exactly the second task:

# GET /tasks?skip=1&limit=1

GET /tasks?skip=1&limit=1 200 [{'done': True, 'id': 2, 'title': 'Email the team'}]

That’s the whole pagination pattern: offset is where the page starts, limit is how big the page is. A client wanting page 3 of 10-row pages would send skip=20&limit=10. Defaults keep simple requests simple while giving callers control when they need it.

Practice Exercises#

Exercise 1: Trace the three save calls#

A teammate writes create_task but forgets session.refresh(task) before returning. The row saves fine, but the response shows "id": null. Why?

Exercise 2: Why model_validate instead of saving the schema directly?#

Why does the create endpoint convert data into a Task with Task.model_validate(data) instead of just calling session.add(data) on the incoming TaskCreate?

Exercise 3: Build a “page 2” request#

Pages are 5 tasks each. Which skip and limit query parameters return the second page (tasks 6 through 10)?

Summary#

You’ve written the first half of CRUD against a real database. Create converts the incoming TaskCreate into a Task row with Task.model_validate, then saves it with the three-step add (stage), commit (write and assign the id), and refresh (reload the saved row) sequence. Read one uses session.get(Task, task_id) and raises a 404 when the row is missing. Read many runs select(Task) with offset and limit for clean pagination. All three plug into the get_session dependency and use response_model to shape their output — and every row now lives on disk.

Key Concepts#

• model_validate — converts a create schema into a savable table row object.
• add / commit / refresh — stage the row, write it (assigning the id), then reload it.
• session.get(Model, id) — fetch one row by primary key, or None if absent.
• 404 guard — raise HTTPException(status_code=404, ...) when a lookup returns None.
• offset / limit — skip and cap rows in a select to paginate a list.

Why This Matters#

Create and read are the operations you’ll write most often in any API — almost every feature boils down to “save this” and “show me that.” The add/commit/refresh rhythm and the get-then-404 guard are patterns you’ll reuse in every project, and pagination is what keeps list endpoints fast and safe as data grows. Master these and the update and delete operations next lesson are small variations on the same moves.

Next Steps#

Continue Building Your Skills#

Your Task Manager can now create and read tasks that persist to a real database. Next you’ll round out CRUD by changing rows that already exist and removing the ones you no longer need — completing the full lifecycle of your data and finishing the database-backed API you set out to build.