FastAPI CRUD with SQLModel and SQLite

From a Dict to a Database

The earlier lessons stored data in a Python dict that was emptied every time the server restarted. This lesson stores it in SQLite, a database kept in a single file, through SQLModel. SQLModel is a library by FastAPI’s author that combines Pydantic models with SQLAlchemy tables, so a single class can be both a table and a validated model.

Install it next to FastAPI with pip install sqlmodel. Outputs come from FastAPI 0.143.0, Pydantic 2.14.1, SQLModel 0.0.48 and SQLAlchemy 2.1.4 on Python 3.12. The apps run under Uvicorn on port 8011 (uvicorn first:app --port 8011); use any free port. If you know SQL, the SQL DDL and SELECT lessons show the statements SQLModel generates for you.

One Table, One Session per Request

Save as first.py:

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


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    age: int | None = None


engine = create_engine("sqlite:///heroes.db")
SQLModel.metadata.create_all(engine)
app = FastAPI()


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


@app.post("/heroes")
def create_hero(hero: Hero, session: Session = Depends(get_session)):
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero


@app.get("/heroes")
def list_heroes(session: Session = Depends(get_session)):
    return session.exec(select(Hero)).all()

$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"name": "Deadpond"}'
{"name":"Deadpond","id":1,"age":null}
$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"name": "Rusty-Man", "age": 48}'
{"name":"Rusty-Man","id":2,"age":48}
$ curl -s http://127.0.0.1:8011/heroes
[{"name":"Deadpond","id":1,"age":null},{"name":"Rusty-Man","id":2,"age":48}]

Four pieces are working together here:

• table=True makes Hero a table, named hero. create_all creates it in heroes.db if it doesn’t exist yet.
• The engine knows how to reach the database. A session is one unit of work on it: you add objects, commit writes them, refresh reloads them so the new id is filled in, and exec(select(Hero)) reads rows back as Hero objects.
• get_session is a dependency that uses yield. FastAPI runs it up to yield before your function and hands your function the session, then finishes the with block afterwards, closing the session. Each request gets its own session.
• The id is None until the database assigns 1, 2, 3 on insert.

Stop the server and start it again, and GET /heroes still returns both heroes, because they’re in the file now.

The Bug in Using the Table as the Body

Hero has an id field, so a client can send one:

$ curl -i http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"id": 1, "name": "Spider-Boy"}'
HTTP/1.1 500 Internal Server Error
date: Tue, 13 Oct 2026 07:19:28 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

The end of the Uvicorn log, after many framework frames:

  File "C:\tmp\fapi-b2\l12\first.py", line 24, in create_hero
    session.commit()
...
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: hero.id
[SQL: INSERT INTO hero (id, name, age) VALUES (?, ?, ?)]
[parameters: (1, 'Spider-Boy', None)]
(Background on this error at: https://sqlalche.me/e/21/gkpj)

The client’s "id": 1 went straight into the INSERT. Row 1 already exists, so SQLite refused it at line 24, the commit, and the exception became a 500. With an unused id, the same request would quietly pick its own primary key. The input model and the table shouldn’t be the same class.

Separate Models for Table, Input, Output and Update

The usual SQLModel layout shares fields through a base class. Save as models.py:

from sqlmodel import Field, SQLModel


class HeroBase(SQLModel):
    name: str = Field(min_length=1, max_length=50)
    age: int | None = Field(default=None, ge=0)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: str | None = Field(default=None, min_length=1, max_length=50)
    age: int | None = Field(default=None, ge=0)

Hero is the only table. HeroCreate is what a client may send: no id. HeroPublic is what the API returns, with id required, since a stored hero always has one. HeroUpdate makes every field optional for PATCH. The Field constraints are the ones from Pydantic Validation.

The database setup goes in db.py. The print calls show when the session opens and closes:

from sqlmodel import Session, SQLModel, create_engine

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


def create_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        print("session opened")
        yield session
        print("session closed")

Create and Read

Save as main.py. The update and delete routes are added below:

from typing import Annotated

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

from db import create_tables, get_session
from models import Hero, HeroCreate, HeroPublic, HeroUpdate

create_tables()
app = FastAPI()
SessionDep = Annotated[Session, Depends(get_session)]


@app.post("/heroes", status_code=201, response_model=HeroPublic)
def create_hero(data: HeroCreate, session: SessionDep):
    hero = Hero.model_validate(data)
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero


@app.get("/heroes", response_model=list[HeroPublic])
def list_heroes(session: SessionDep, offset: int = 0, limit: int = Query(default=10, le=100)):
    return session.exec(select(Hero).offset(offset).limit(limit)).all()


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def get_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

Delete the old heroes.db first, then start uvicorn main:app --port 8011:

$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"id": 1, "name": "Deadpond"}'
{"name":"Deadpond","age":null,"id":1}
$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"id": 1, "name": "Spider-Boy", "age": 16}'
{"name":"Spider-Boy","age":16,"id":2}
$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"name": "", "age": -1}'
{"detail":[{"type":"string_too_short","loc":["body","name"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}},{"type":"greater_than_equal","loc":["body","age"],"msg":"Input should be greater than or equal to 0","input":-1,"ctx":{"ge":0}}]}
$ curl -s http://127.0.0.1:8011/heroes
[{"name":"Deadpond","age":null,"id":1},{"name":"Spider-Boy","age":16,"id":2}]
$ curl -s http://127.0.0.1:8011/heroes/2
{"name":"Spider-Boy","age":16,"id":2}
$ curl -i http://127.0.0.1:8011/heroes/99
HTTP/1.1 404 Not Found
date: Tue, 13 Oct 2026 07:19:46 GMT
server: uvicorn
content-length: 27
content-type: application/json

{"detail":"Hero not found"}

The second request sent "id": 1 again and got id 2: HeroCreate has no id field, so the key was dropped before it reached the database. Hero.model_validate(data) copies the validated input into a table object. The constraints on HeroBase gave a proper 422 instead of a bad row. SessionDep is an Annotated alias, so each route says session: SessionDep instead of repeating Depends(get_session).

For a single row, session.get(Hero, hero_id) looks it up by primary key and returns None when there is none. Turning that None into a 404 is your job; without the check, response_model=HeroPublic would fail on None and the client would get a 500. The list route caps limit at 100, so a client can’t ask for the whole table at once.

Update and Delete

Add these two routes to the end of main.py:

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, data: HeroUpdate, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero.sqlmodel_update(data.model_dump(exclude_unset=True))
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero


@app.delete("/heroes/{hero_id}", status_code=204)
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()

Continuing with the same server:

$ curl -s -X PATCH http://127.0.0.1:8011/heroes/2 -H 'content-type: application/json' -d '{"age": 17}'
{"name":"Spider-Boy","age":17,"id":2}
$ curl -s -X PATCH http://127.0.0.1:8011/heroes/2 -H 'content-type: application/json' -d '{"age": null}'
{"name":"Spider-Boy","age":null,"id":2}
$ curl -i -X DELETE http://127.0.0.1:8011/heroes/1
HTTP/1.1 204 No Content
date: Tue, 13 Oct 2026 07:19:47 GMT
server: uvicorn
content-type: application/json


$ curl -s -X DELETE http://127.0.0.1:8011/heroes/1
{"detail":"Hero not found"}
$ curl -s 'http://127.0.0.1:8011/heroes?limit=500'
{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 100","input":"500","ctx":{"le":100}}]}
$ curl -s http://127.0.0.1:8011/heroes
[{"name":"Spider-Boy","age":null,"id":2}]

The first PATCH sent only age, and name survived. That’s exclude_unset=True: model_dump returns only the keys the client sent, here {"age": 17}, and sqlmodel_update copies just those onto the row. Without it, every field the client left out would be overwritten with its None default. The second PATCH sent null on purpose, and the age was cleared. That’s how a client erases a value.

DELETE answers 204 with no body. Deleting the same hero again gives 404, so a client can tell “deleted” from “wasn’t there”.

What the Session Log Shows

Here is the Uvicorn log from those requests, unbuffered (PYTHONUNBUFFERED=1) so the print lines appear as they happen:

session opened
INFO:     127.0.0.1:63963 - "POST /heroes HTTP/1.1" 201 Created
session closed
session opened
INFO:     127.0.0.1:63964 - "POST /heroes HTTP/1.1" 201 Created
session closed
session opened
INFO:     127.0.0.1:63965 - "POST /heroes HTTP/1.1" 422 Unprocessable Entity
session opened
INFO:     127.0.0.1:63966 - "GET /heroes HTTP/1.1" 200 OK
session closed
session opened
INFO:     127.0.0.1:63967 - "GET /heroes/2 HTTP/1.1" 200 OK
session closed
session opened
INFO:     127.0.0.1:63968 - "GET /heroes/99 HTTP/1.1" 404 Not Found

Two things stand out. On success, “session closed” prints after the response line: FastAPI runs the code after yield once the response has gone out. After the 422 and the 404 there is no “session closed” at all. FastAPI throws the exception into the generator at yield, so the print after it never runs. The session itself is still closed, because the with block exits either way, but any cleanup written as a plain line after yield is skipped. Practice task 1 fixes that. The 422 line also shows that the session was opened even though the body was invalid: dependencies are resolved alongside the request data, so don’t do expensive work in a dependency on the assumption that the request is valid.

A Note on check_same_thread

Many FastAPI and SQLModel examples create the engine with connect_args={"check_same_thread": False}, because SQLite’s Python driver refuses by default to use a connection from a thread other than the one that created it, and FastAPI runs def routes on a thread pool. With SQLAlchemy 2.x it’s no longer needed. Save as engine_args.py:

from sqlmodel import create_engine

engine = create_engine("sqlite:///heroes.db")
print(engine.pool.__class__.__name__)
print(engine.dialect.create_connect_args(engine.url))

$ python engine_args.py
QueuePool
(['C:\\tmp\\fapi-b2\\l12\\heroes.db'], {'check_same_thread': False})

For a file database, SQLAlchemy already passes check_same_thread=False and pools connections. Writing the option yourself is harmless, just redundant.

Practice

1. Change get_session in db.py so that “session closed” prints for every request, including 404s and 422s.

from sqlmodel import Session, SQLModel, create_engine

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


def create_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        print("session opened")
        try:
            yield session
        finally:
            print("session closed")

$ curl -s http://127.0.0.1:8011/heroes/99
{"detail":"Hero not found"}
$ curl -s http://127.0.0.1:8011/heroes -H 'content-type: application/json' -d '{"name": ""}'
{"detail":[{"type":"string_too_short","loc":["body","name"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}}]}
$ curl -s http://127.0.0.1:8011/heroes
[]

session opened
session closed
INFO:     127.0.0.1:63997 - "GET /heroes/99 HTTP/1.1" 404 Not Found
session opened
session closed
INFO:     127.0.0.1:63998 - "POST /heroes HTTP/1.1" 422 Unprocessable Entity
session opened
INFO:     127.0.0.1:63999 - "GET /heroes HTTP/1.1" 200 OK
session closed

2. Make hero names unique in the database. A duplicate name should get 409 Conflict with a clear message instead of a 500. Keep it to one file with a name query parameter.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Session, SQLModel, create_engine


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(unique=True)


engine = create_engine("sqlite:///unique.db")
SQLModel.metadata.create_all(engine)
app = FastAPI()


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


@app.post("/heroes", status_code=201)
def create_hero(name: str, session: Annotated[Session, Depends(get_session)]) -> Hero:
    hero = Hero(name=name)
    session.add(hero)
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        raise HTTPException(status_code=409, detail=f"Hero '{name}' already exists")
    session.refresh(hero)
    return hero

$ curl -s -X POST 'http://127.0.0.1:8011/heroes?name=Deadpond'
{"name":"Deadpond","id":1}
$ curl -i -X POST 'http://127.0.0.1:8011/heroes?name=Deadpond'
HTTP/1.1 409 Conflict
date: Tue, 13 Oct 2026 07:20:41 GMT
server: uvicorn
content-length: 43
content-type: application/json

{"detail":"Hero 'Deadpond' already exists"}

3. Add optional q (name contains) and min_age filters to GET /heroes, sorted by name.

@app.get("/heroes", response_model=list[HeroPublic])
def list_heroes(session: SessionDep, q: str | None = None, min_age: int | None = None):
    query = select(Hero).order_by(Hero.name)
    if q is not None:
        query = query.where(col(Hero.name).contains(q))
    if min_age is not None:
        query = query.where(Hero.age >= min_age)
    return session.exec(query).all()

$ curl -s 'http://127.0.0.1:8011/heroes'
[{"name":"Deadpond","age":null,"id":3},{"name":"Rusty-Man","age":48,"id":2},{"name":"Spider-Boy","age":16,"id":1}]
$ curl -s 'http://127.0.0.1:8011/heroes?q=man'
[{"name":"Rusty-Man","age":48,"id":2}]
$ curl -s 'http://127.0.0.1:8011/heroes?min_age=18'
[{"name":"Rusty-Man","age":48,"id":2}]