CRUD Operations with a Real Database¶

🎯 What You’ll Learn¶

How to use SQLModel’s session system to interact with the database
How to create, read, update, and delete records using SQLModel
How to integrate database logic into your FastAPI routers
How to maintain clean separation between routing and persistence logic

🧠 Step 1: Import the Session System¶

📄 db.py (update this file)

from sqlmodel import SQLModel, create_engine, Session

sqlite_url = "sqlite:///./tasks.db"
engine = create_engine(sqlite_url, echo=True)

def get_session():
    return Session(engine)

Session(engine) creates a database session
get_session() is a helper function to reuse sessions cleanly

📦 Step 2: Update Your Router to Use Sessions¶

📄 routers/tasks.py

from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import select
from models.task import Task
from db import get_session
from sqlmodel import Session

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

✅ Create a Task (POST /tasks)¶

@router.post("/", status_code=201)
def create_task(task: Task, session: Session = Depends(get_session)):
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

session.add() queues the task for insertion
session.commit() writes it to the database
session.refresh() reloads the task with its new ID

📖 Read All Tasks (GET /tasks)¶

@router.get("/")
def get_all_tasks(session: Session = Depends(get_session)):
    tasks = session.exec(select(Task)).all()
    return tasks

select(Task) builds a query
session.exec(...).all() executes and returns all results

📖 Read a Single Task (GET /tasks/{task_id})¶

@router.get("/{task_id}")
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

session.get(Task, id) fetches by primary key

✏️ Update a Task (PUT /tasks/{task_id})¶

@router.put("/{task_id}")
def update_task(task_id: int, updated_task: Task, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    task.title = updated_task.title
    task.description = updated_task.description
    task.completed = updated_task.completed
    session.commit()
    session.refresh(task)
    return task

Fetch the task
Update fields manually
Commit and refresh

❌ Delete a Task (DELETE /tasks/{task_id})¶

@router.delete("/{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()

session.delete() removes the record
session.commit() finalizes the deletion

📚 Swagger UI Still Works¶

Visit http://127.0.0.1:8000/docs and you’ll see:

All endpoints grouped under “Tasks”
Request and response schemas
Interactive testing with real database persistence

🧠 Recap¶

You now have a production-grade CRUD API backed by a real database:

Method	Endpoint	Action
POST	`/tasks`	Create a task
GET	`/tasks`	Read all tasks
GET	`/tasks/{id}`	Read one task
PUT	`/tasks/{id}`	Update a task
DELETE	`/tasks/{id}`	Delete a task

All operations use SQLModel sessions and persist data to tasks.db.

🧪 Practice Challenge¶

Add a new endpoint:

GET /tasks/completed → returns only tasks where completed=True
Use a select(Task).where(Task.completed == True) query
Test it in Swagger UI