Hands-On: Database with SQLModel & Dependency Injection

Why Do We Need a Real Database?

Our API from the previous article already has full CRUD with decent validation, but its data still lives in a Python list — lost every time the server restarts, and unusable if more than one server process is running. In this article we connect the app to a real database (SQLite, via SQLModel), and finally cover the dependency injection concept we only briefly touched on in Part 2.

SQLModel — One Model for Both Validation and the Database

SQLModel is a library, built by the creator of FastAPI himself, that combines Pydantic (data validation) with SQLAlchemy (ORM/database). Its main benefit: we can define a single class that acts as both a Pydantic schema AND a database table — no need to write two separate definitions that easily drift out of sync.

pip install sqlmodel

from typing import Optional
from sqlmodel import Field, SQLModel


class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    done: bool = False
    priority: Priority = Priority.medium

Notice table=True — this is what sets Task apart from a regular Pydantic model: this class also represents a real SQL table. id is typed Optional[int] with a default of None because its value only gets filled in by the database (auto-increment) after the row is saved, not by us.

Engine and Session — How SQLModel Talks to the Database

Two basic concepts before we can query anything:

# database.py
from sqlmodel import Session, SQLModel, create_engine

DATABASE_URL = "sqlite:///./tasks.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


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

connect_args={"check_same_thread": False} is specifically needed for SQLite, because by default SQLite only allows access from the thread that created the connection — while FastAPI can process requests from different threads. For other databases (PostgreSQL, MySQL), this line isn't needed.

Notice get_session uses yield, not return. This is the generator pattern — we'll cover why it matters in the next section.

Dependency Injection — Delivering on the Promise From Part 2

Now it's time to cover Depends() in full. Dependency injection is how FastAPI provides something a path operation function needs — in our case, a database Session — without the endpoint itself needing to know how to create it.

from fastapi import Depends
from sqlmodel import Session

from .database import get_session


@app.get("/api/tasks")
def list_tasks(session: Session = Depends(get_session)):
    ...

Once FastAPI sees Depends(get_session), it will: call get_session(), run it up to the yield (at that point, session gets handed to the list_tasks function), then once list_tasks finishes (whether it succeeds or errors), FastAPI resumes get_session right after the yield — which in this case closes the session via with Session(engine) as session.

Why is this better than creating a session manually in every endpoint?

Cleaning Up the Project Structure

Our project now has a few different concerns (models, database connection, endpoints). It's time to split it out of a single main.py file into a more structured package:

task-tracker-api/
├── app/
│   ├── main.py             # FastAPI app setup, lifespan, health check
│   ├── database.py         # engine, get_session
│   ├── models.py            # Task (SQLModel, table=True), Priority
│   ├── schemas.py           # TaskCreate, TaskUpdate, TaskRead
│   └── routers/
│       └── tasks.py         # all /api/tasks/* endpoints
└── tests/
    └── test_tasks.py

APIRouter is used to group related endpoints into a file separate from main.py:

# routers/tasks.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from app.database import get_session
from app.models import Task
from app.schemas import TaskCreate, TaskRead

router = APIRouter(prefix="/api/tasks", tags=["tasks"])


@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: Session = Depends(get_session)):
    task = Task(title=payload.title, priority=payload.priority)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

prefix="/api/tasks" means we don't need to repeat /api/tasks on every decorator — just @router.post("") for POST /api/tasks, @router.get("/{task_id}") for GET /api/tasks/{task_id}, and so on. The session.add() → session.commit() → session.refresh() pattern is the standard SQLModel/SQLAlchemy insert flow: add the object to the session, save it to the database, then re-fetch it (so the database-generated id gets filled in).

For filtered queries, SQLModel uses select(), similar to plain SQL but still type-safe:

@router.get("", response_model=list[TaskRead])
def list_tasks(done: bool | None = None, session: Session = Depends(get_session)):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    return session.exec(query).all()

Finally, main.py just wires everything together, and makes sure the database tables get created on app startup via a lifespan handler:

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import tasks


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Task Tracker API", lifespan=lifespan)


@app.get("/health")
def health_check():
    return {"status": "ok"}


app.include_router(tasks.router)

lifespan is FastAPI's modern way of running code at app startup (before yield) and app shutdown (after yield) — in our case, just making sure the tables exist before the first request comes in.

Running and Testing

pip install -r requirements.txt
uvicorn app.main:app --reload

Try creating a few tasks via /docs or curl, then restart the server. Unlike before, the tasks you created are still there — because the data now lives in the tasks.db file, not in process memory that disappears the moment the process stops.

Summary