Lesson 5.3 · Databases

CRUD with Sessions

Connect SQLModel to FastAPI: one session per request through a yield dependency, tables made at startup with lifespan, and the five to-do endpoints - create, list with paging, read, PATCH and delete - saved to a file that survives a restart. Includes a real 500 bug from PATCH {"title": null}, its fix, and tests on an in-memory database.

Intermediate35 min

What you will be able to do

Give every request its own session with a yield dependency
Create the tables when the app starts, with lifespan
Write create, list, read, update and delete endpoints with SQLModel
Update only the fields a client sent - and refuse null for required columns
Return table objects through public response models
Test with an in-memory database through dependency_overrides

The idea, in plain English

Lesson 5.2 used SQLModel from a script. In an API, each request needs its own session: opened when the request starts, closed when it ends, never shared between two requests. That is exactly a dependency with yield (Lesson 4.3).

The endpoints then follow one pattern. Input comes in as a data model (TodoCreate, TodoUpdate), the endpoint works with the table model (Todo) through the session, and the answer goes out through a public model (TodoPublic). The database checks its own rules too - and when the two disagree, you get a 500.

We rebuild the to-do API on SQLite, test every endpoint, restart the server to prove the data stays, fix a real bug, and write tests on a throw-away database. FastAPI 0.143.0, SQLModel 0.0.48, SQLAlchemy 2.1.4, pytest 9.1.1.

Worked example: The to-do API from Module 1, now saved in todos.db - tested with TestClient, with curl across a server restart, and with pytest on an in-memory database.

workflowOne request, one sessionstep 1 / 3

request

PATCH /todos/1

get_session()

with Session(engine)

update_todo()

get, change, commit

todos.db

UPDATE todo ...

✓

React Flow

1 - Change one field

The client sends {"done": true}. exclude_unset keeps only that field; the title stays.

answer

200 done: true, title: Buy milk

The session is opened for the request and closed after it. Real results from our to-do API.

Words you will see in this lesson

A few words about CRUD APIs.

Small dictionary

CRUDCreate, Read, Update, Delete - the four basic jobs on data.

LifespanCode that runs once when the app starts (and once when it stops).

Offset / limitSkip this many rows, then return at most this many - paging.

In-memory databaseA SQLite database that lives only in RAM - empty every time.

StaticPoolA pool that keeps one shared connection - needed for an in-memory test database.

An everyday example: a library counter

Each visitor gets the counter for their visit - one session per request.
New books are checked (TodoCreate) before they get a shelf number (the id).
A change form has only the boxes you fill in (PATCH) - an empty box means "no change", not "erase it".
The register is a real book - close the library tonight, and it is still there tomorrow.

Example 1 - the models

The four models from Lesson 5.2, plus TodoUpdate for PATCH:

TodoUpdate - every field optional, so the client can send only what changes.
The validator not_null refuses an explicit null. It runs only for fields the client sent; a missing field keeps its default and is never checked. Why we need it is in "A real bug" below.

Example 1 - models.py

from datetime import datetime, timezone

from pydantic import field_validator
from sqlmodel import Field, SQLModel


class TodoBase(SQLModel):
    title: str = Field(min_length=1, max_length=100)
    done: bool = False


class Todo(TodoBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class TodoCreate(TodoBase):
    pass


class TodoUpdate(SQLModel):                          # every field optional - for PATCH
    title: str | None = Field(default=None, min_length=1, max_length=100)
    done: bool | None = None

    @field_validator("title", "done")
    @classmethod
    def not_null(cls, value):                        # runs only for fields the client SENT
        if value is None:
            raise ValueError("may be left out, but not null")
        return value


class TodoPublic(TodoBase):
    id: int
    created_at: datetime

Example 2 - the API

Read main.py from the top:

One engine for the app. get_session() opens a Session, yields it, and closes it - one per request.
SessionDep uses scope="function", so the session is finished before the answer is sent (Lesson 4.3).
lifespan runs create_all once, when the server starts - not on every request.
create_todo: data (TodoCreate) -> Todo.model_validate -> add, commit, refresh. response_model=TodoPublic shapes the answer; 201 Created.
list_todos: select, an optional filter on done, order_by, offset and limit with rules.
get_todo_or_404 is a helper used by read, PATCH and delete.
update_todo: model_dump(exclude_unset=True) gives only the sent fields; sqlmodel_update copies them onto the row.
delete_todo: session.delete, commit, 204.

Example 2 - main.py

from contextlib import asynccontextmanager
from typing import Annotated

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

from models import Todo, TodoCreate, TodoPublic, TodoUpdate

engine = create_engine("sqlite:///todos.db")


def get_session():
    with Session(engine) as session:                 # one session per request
        yield session


SessionDep = Annotated[Session, Depends(get_session, scope="function")]   # done before the answer (Lesson 4.3)


@asynccontextmanager
async def lifespan(app: FastAPI):
    SQLModel.metadata.create_all(engine)             # once, when the server starts
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/todos", response_model=TodoPublic, status_code=201)
def create_todo(data: TodoCreate, session: SessionDep):
    todo = Todo.model_validate(data)
    session.add(todo)
    session.commit()
    session.refresh(todo)
    return todo


@app.get("/todos", response_model=list[TodoPublic])
def list_todos(session: SessionDep,
               offset: Annotated[int, Query(ge=0)] = 0,
               limit: Annotated[int, Query(ge=1, le=100)] = 20,
               done: bool | None = None):
    query = select(Todo)
    if done is not None:
        query = query.where(Todo.done == done)
    return session.exec(query.order_by(Todo.id).offset(offset).limit(limit)).all()


def get_todo_or_404(session: Session, todo_id: int) -> Todo:
    todo = session.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo


@app.get("/todos/{todo_id}", response_model=TodoPublic)
def read_todo(todo_id: int, session: SessionDep):
    return get_todo_or_404(session, todo_id)


@app.patch("/todos/{todo_id}", response_model=TodoPublic)
def update_todo(todo_id: int, data: TodoUpdate, session: SessionDep):
    todo = get_todo_or_404(session, todo_id)
    changes = data.model_dump(exclude_unset=True)    # only what the client sent
    todo.sqlmodel_update(changes)
    session.add(todo)
    session.commit()
    session.refresh(todo)
    return todo


@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, session: SessionDep):
    todo = get_todo_or_404(session, todo_id)
    session.delete(todo)
    session.commit()

try_crud.py

from fastapi.testclient import TestClient

from main import app

with TestClient(app, raise_server_exceptions=False) as c:   # "with" runs the lifespan
    def show(label, r):
        body = r.text if r.text else "(empty)"
        print(f"{label:<20} -> {r.status_code} {body}")

    show("create 1", c.post("/todos", json={"title": "Buy milk"}))
    show("create 2", c.post("/todos", json={"title": "Pay rent"}))
    show("create, empty", c.post("/todos", json={"title": ""}))
    show("read 1", c.get("/todos/1"))
    show("patch done only", c.patch("/todos/1", json={"done": True}))
    show("patch title null", c.patch("/todos/1", json={"title": None}))
    show("list done=false", c.get("/todos", params={"done": False}))
    show("delete 2", c.delete("/todos/2"))
    show("read 2", c.get("/todos/2"))
    show("list", c.get("/todos"))

Reading the output

created_at is shortened to "..." in the output below.

create - 201 and the id from the database (1, then 2).
create with an empty title - 422 from TodoCreate. The database was never asked.
patch done only - done changed, the title stayed "Buy milk".
patch title null - 422 with our message.
list done=false - only "Pay rent".
delete - 204; reading it again - 404 from our helper.

Output

create 1             -> 201 {"title":"Buy milk","done":false,"id":1,"created_at":"..."}
create 2             -> 201 {"title":"Pay rent","done":false,"id":2,"created_at":"..."}
create, empty        -> 422 {"detail":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}}]}
read 1               -> 200 {"title":"Buy milk","done":false,"id":1,"created_at":"..."}
patch done only      -> 200 {"title":"Buy milk","done":true,"id":1,"created_at":"..."}
patch title null     -> 422 {"detail":[{"type":"value_error","loc":["body","title"],"msg":"Value error, may be left out, but not null","input":null,"ctx":{"error":{}}}]}
list done=false      -> 200 [{"title":"Pay rent","done":false,"id":2,"created_at":"..."}]
delete 2             -> 204 (empty)
read 2               -> 404 {"detail":"Todo not found"}
list                 -> 200 [{"title":"Buy milk","done":true,"id":1,"created_at":"..."}]

A real bug: PATCH with null

Our first TodoUpdate had no validator - only title: str | None = None:

PATCH {"title": null} passed TodoUpdate - "str | None" allows None.
exclude_unset kept it, because the client DID send title.
The UPDATE reached the database, and the NOT NULL rule refused it: "NOT NULL constraint failed: todo.title".
The client got 500 Internal Server Error for a simple input mistake.
The fix is the not_null validator: now it is a clear 422, and the database is never asked.
Rule: "optional in PATCH" means "may be left out", not "may be null". Check every optional PATCH field against its column.

Output - the first version (only the different line)

patch title null     -> 500 Internal Server Error

server log:
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: todo.title
[SQL: UPDATE todo SET title=? WHERE todo.id = ?]
[parameters: (None, 1)]

The data survives a restart

We ran the app with uvicorn, added a to-do with curl, stopped the server and started it again:

"Water plants" was still there, next to the to-do from the earlier test.
create_all at the second start did nothing - the table already existed (Lesson 5.2).
Compare Module 1, where a restart emptied the dict.

Output - curl, restart, curl

$ curl -s -X POST localhost:8351/todos -H "Content-Type: application/json" -d '{"title": "Water plants"}'
{"title":"Water plants","done":false,"id":2,"created_at":"..."}
(server stopped and started again)
$ curl -s localhost:8351/todos
[{"title":"Buy milk","done":true,"id":1,"created_at":"..."},{"title":"Water plants","done":false,"id":2,"created_at":"..."}]

check_same_thread: needed or not?

Many tutorials add connect_args={"check_same_thread": False} for SQLite with FastAPI. We tested it:

Without it, 100 POSTs in parallel (20 at a time) all answered 201.
The reason: SQLAlchemy 2.x already turns check_same_thread off for a SQLite FILE database. create_engine("sqlite:///x.db") showed {"check_same_thread": False} in its connect args.
For an IN-MEMORY database it does not. Our tests failed with "SQLite objects created in a thread can only be used in that same thread" until we added it.
So: not needed for todos.db; needed for "sqlite://" in tests (next section).

Tests on an in-memory database

Tests should not touch todos.db. We override get_session (Lesson 4.1):

create_engine("sqlite://") is an in-memory database - empty for every test.
StaticPool keeps one shared connection, so the tables made by create_all are still there for the requests.
connect_args={"check_same_thread": False}, because the requests run in other threads.
Both tests passed, and the first id was 1 in each - a fresh database. todos.db still had its one row.

test_todos.py

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from main import app, get_session


@pytest.fixture
def client():
    engine = create_engine("sqlite://",                     # in memory: empty for every test
                           connect_args={"check_same_thread": False},
                           poolclass=StaticPool)            # one shared connection
    SQLModel.metadata.create_all(engine)

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

    app.dependency_overrides[get_session] = get_test_session
    yield TestClient(app)
    app.dependency_overrides.clear()


def test_create_and_read(client):
    r = client.post("/todos", json={"title": "Test me"})
    assert r.status_code == 201
    todo_id = r.json()["id"]
    assert todo_id == 1                              # a fresh database every time
    assert client.get(f"/todos/{todo_id}").json()["title"] == "Test me"


def test_patch_null_is_refused(client):
    client.post("/todos", json={"title": "Keep my title"})
    assert client.patch("/todos/1", json={"title": None}).status_code == 422
    assert client.get("/todos/1").json()["title"] == "Keep my title"

Output - python -m pytest -q test_todos.py

..                                                                       [100%]
2 passed in 0.23s

CRUD with SQLModel at a glance

Session per request

A yield dependency.

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

Create

Then refresh for the id.

todo = Todo.model_validate(data)
session.add(todo); session.commit(); session.refresh(todo)

List

Filter, order, page.

session.exec(select(Todo).where(...).order_by(Todo.id).offset(o).limit(l)).all()

Read

None if missing.

session.get(Todo, todo_id)

Update

Only sent fields.

todo.sqlmodel_update(data.model_dump(exclude_unset=True))

Delete

Then commit.

session.delete(todo); session.commit()

Try it yourself

The code does not change. Swap the content string and the program does something else entirely.

See the SQL

“Add echo=True to create_engine and call PATCH /todos/1 with {"done": false}. Which UPDATE is sent?”

Use /docs

“Run fastapi dev main.py and use /docs to create, list and delete to-dos. Look at todos.db with sqlite3.”

A search

“Add q: str | None = None to list_todos and filter with Todo.title.contains(q).”

Count

“Add GET /todos/count that returns the number of rows (select(func.count()).select_from(Todo)).”

What usually goes wrong

One session for the whole app

Sessions are not safe to share between requests. Make one per request in a dependency.

create_all on every request

Run it once at startup in lifespan (and use migrations later - Lesson 5.5).

Allowing null in PATCH for a NOT NULL column

Our PATCH {"title": null} became a 500. Refuse explicit null in the update model.

Returning the table model without a response model

Every column is sent - fine today, a leak when you add a secret column. Use response_model=TodoPublic.

Tests on the real database file

Override get_session with an in-memory engine.

Practice

Write these yourself before opening anything. Getting them wrong first is most of how this sticks.

Add PUT /todos/{todo_id} that replaces title and done completely (both required, TodoCreate as the body), and POST /todos/{todo_id}/toggle that flips done. Write pytest tests for both on the in-memory database, including a 404 for an unknown id.

Show hint

PUT: todo.sqlmodel_update(data.model_dump()). Toggle: todo.done = not todo.done. Reuse get_todo_or_404.

Key points

One session per request, from a yield dependency.
lifespan runs startup work like create_all once.
Data models in, table model inside, public model out.
PATCH: model_dump(exclude_unset=True) + sqlmodel_update - and refuse explicit null for required columns.
SQLAlchemy 2.x sets check_same_thread=False for SQLite files, not for in-memory databases.
Tests override get_session with sqlite:// and StaticPool.

Quick check before you move on

What did PATCH {"title": null} return in the first version?

500 - the database refused NULL for title.

Why does exclude_unset matter for PATCH?

Without it, fields the client did not send would be set to their defaults (None).

Was "Water plants" still there after the restart?

Yes - it is stored in todos.db.

Why does the test engine need StaticPool?

An in-memory database lives in one connection; StaticPool shares that one connection.

Interview questions

How do you manage database sessions in a FastAPI app?

Create one engine at startup and one session per request via a yield dependency, closing it in all cases; commit inside the endpoint or with a function-scoped dependency so failures become error responses; never share sessions across requests or threads.

How do you implement PATCH correctly?

Use an update model with optional fields, apply only fields that were sent (exclude_unset), validate that explicit nulls are allowed for the target columns, and return the refreshed resource.

Quiz

1.
Why does not_null not run when a client leaves title out?
2.
Why was check_same_thread not needed for todos.db?
3.
What does refresh do after commit in create_todo?

Comments

Loading comments...