Learning Goals
3 min- Write the four CRUD verbs in SQL and wire each to a Python function.
- Use a single
get_conn()helper to share a connection. - Return the new
lastrowidafter an INSERT. - Check
rowcountafter UPDATE/DELETE so the caller knows if anything happened.
Warm-Up · The Schema
5 minCREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, done INTEGER DEFAULT 0, -- 0 = open, 1 = done created_at TEXT DEFAULT CURRENT_TIMESTAMP );
AUTOINCREMENTguarantees ids are never reused.NOT NULLsaystitlecan't be missing.DEFAULTfills the value when you don't supply one.CURRENT_TIMESTAMPstamps the row with UTC time at INSERT.
Wrap each SQL verb in one function. The rest of your program never writes SQL inline. That separation makes UI changes trivial later.
New Concept · One Function Per Verb
14 minThe shared connection
import sqlite3 from pathlib import Path DB = Path("tasks.db") def get_conn(): con = sqlite3.connect(DB) con.row_factory = sqlite3.Row return con def init(): with get_conn() as con: con.execute(""" CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, done INTEGER DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) """)
Create
def add_task(title): with get_conn() as con: cur = con.execute("INSERT INTO tasks (title) VALUES (?)", (title,)) return cur.lastrowid # the auto-generated id
Read
def list_tasks(done=None): sql = "SELECT * FROM tasks" params = () if done is not None: sql += " WHERE done = ?" params = (1 if done else 0,) sql += " ORDER BY id" with get_conn() as con: return [dict(r) for r in con.execute(sql, params)] def find_task(task_id): with get_conn() as con: row = con.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() return dict(row) if row else None
Update
def mark_done(task_id): with get_conn() as con: cur = con.execute("UPDATE tasks SET done = 1 WHERE id = ?", (task_id,)) return cur.rowcount # 1 if a row changed, 0 if id didn't exist def rename_task(task_id, new_title): with get_conn() as con: return con.execute( "UPDATE tasks SET title = ? WHERE id = ?", (new_title, task_id), ).rowcount
Delete
def delete_task(task_id): with get_conn() as con: return con.execute( "DELETE FROM tasks WHERE id = ?", (task_id,) ).rowcount
The pattern
Every write returns either lastrowid (INSERT) or rowcount (UPDATE/DELETE). That lets the caller respond — "saved as id 5" or "no such task". Reads return Python lists or dicts, never bare cursors — so the rest of the program is database-agnostic.
Worked Example · tasks_cli.py
12 minGlue the CRUD helpers into a menu-driven CLI.
# tasks_cli.py — full CRUD menu from tasks_db import init, add_task, list_tasks, mark_done, delete_task init() def show(tasks): if not tasks: print(" (none)") return for t in tasks: tick = "✅" if t["done"] else "⬜" print(f" {tick} #{t['id']:<3} {t['title']} (added {t['created_at']})") MENU = """ 1) list open 2) list all 3) add 4) mark done 5) delete 6) quit """ while True: print(MENU) choice = input("> ").strip() if choice == "1": show(list_tasks(done=False)) elif choice == "2": show(list_tasks()) elif choice == "3": title = input(" task: ").strip() if title: new_id = add_task(title) print(f" ✅ added #{new_id}") elif choice == "4": tid = int(input(" id to mark done: ")) if mark_done(tid): print(" ✅ done") else: print(" ❌ no such task") elif choice == "5": tid = int(input(" id to delete: ")) if delete_task(tid): print(" 🗑️ deleted") else: print(" ❌ no such task") elif choice == "6": break
Sample run
1) list open 2) list all 3) add 4) mark done 5) delete 6) quit > 3 task: Buy roti canai ✅ added #1 > 3 task: Submit homework ✅ added #2 > 4 id to mark done: 1 ✅ done > 1 ⬜ #2 Submit homework (added 2026-05-28 06:42:11)
Read the diff
The CLI knows nothing about SQL. Replace tasks_db.py with a Postgres backend tomorrow and the menu doesn't need to change. That's the "repository" pattern — and you just used it.
Try It Yourself
13 minAdd a function search_tasks(keyword) that returns tasks whose title contains the keyword (use LIKE).
Hint
def search_tasks(keyword): with get_conn() as con: rows = con.execute( "SELECT * FROM tasks WHERE title LIKE ?", (f"%{keyword}%",), ).fetchall() return [dict(r) for r in rows]
Read a list of task titles from a file (one per line) and add them with one executemany call.
Hint
from pathlib import Path titles = [t.strip() for t in Path("seed.txt").read_text().splitlines() if t.strip()] with get_conn() as con: con.executemany("INSERT INTO tasks (title) VALUES (?)", [(t,) for t in titles])
Write stats() that returns a dict {"open": n_open, "done": n_done, "total": n} in a single connection.
Hint
def stats(): with get_conn() as con: row = con.execute(""" SELECT SUM(done = 0) AS open, SUM(done = 1) AS done, COUNT(*) AS total FROM tasks """).fetchone() return dict(row)
SQLite treats booleans as integers, so SUM(done = 0) counts open rows.
Mini-Challenge · Soft Delete
8 minReal apps rarely DELETE — they "soft-delete" by marking a row. Add a deleted_at column. Update the read functions to skip soft-deleted rows by default. Add restore_task(id) that clears the flag.
Show one possible solution
# tasks_db.py changes # init(): con.execute("ALTER TABLE tasks ADD COLUMN deleted_at TEXT") def soft_delete(task_id): with get_conn() as con: return con.execute( "UPDATE tasks SET deleted_at = CURRENT_TIMESTAMP WHERE id = ?", (task_id,), ).rowcount def restore_task(task_id): with get_conn() as con: return con.execute( "UPDATE tasks SET deleted_at = NULL WHERE id = ?", (task_id,), ).rowcount def list_tasks(done=None, include_deleted=False): sql = "SELECT * FROM tasks WHERE 1=1" params = [] if not include_deleted: sql += " AND deleted_at IS NULL" if done is not None: sql += " AND done = ?" params.append(1 if done else 0) sql += " ORDER BY id" with get_conn() as con: return [dict(r) for r in con.execute(sql, params)]
Note ALTER TABLE ADD COLUMN is one of the few schema migrations SQLite handles cheaply. Real production uses migration libraries (alembic, sqlite-utils).
Recap
3 minCRUD is the contract: every database app exposes Create, Read, Update, Delete. Wrap each verb in a function; never write SQL outside the data module. Use lastrowid for INSERTs and rowcount for UPDATE/DELETE so callers can react. Tomorrow we ship — a full library management system.
Homework
4 minBuild notebook.py — a private journal app:
- Schema:
entries(id, title, body, created_at). - CRUD functions for each verb.
- CLI with: list, add, edit, search, delete.
- BONUS: print the latest 5 entries on the home screen.
# notebook_db.py import sqlite3 from pathlib import Path DB = Path("notebook.db") def get_conn(): con = sqlite3.connect(DB); con.row_factory = sqlite3.Row return con def init(): with get_conn() as con: con.execute(""" CREATE TABLE IF NOT EXISTS entries ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, body TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) """) def add(title, body): with get_conn() as con: return con.execute("INSERT INTO entries (title, body) VALUES (?,?)", (title, body)).lastrowid def latest(n=5): with get_conn() as con: return [dict(r) for r in con.execute( "SELECT * FROM entries ORDER BY id DESC LIMIT ?", (n,))] def search(term): with get_conn() as con: return [dict(r) for r in con.execute( "SELECT * FROM entries WHERE title LIKE ? OR body LIKE ?", (f"%{term}%", f"%{term}%"))] def edit(eid, title, body): with get_conn() as con: return con.execute("UPDATE entries SET title=?, body=? WHERE id=?", (title, body, eid)).rowcount def delete(eid): with get_conn() as con: return con.execute("DELETE FROM entries WHERE id=?", (eid,)).rowcount
Non-negotiables: all SQL behind functions, prepared statements only, rowcount/lastrowid returned, CLI never touches sqlite3 directly.