114. Sqlite3 Module

🗃️ A contacts app stores and queries data in SQLite. The developer built INSERT queries using f-strings (SQL injection risk!), forgot to call conn.commit() so data is lost, and used fetchone() where fetchall() is needed to retrieve multiple rows.

💡 Fun fact: SQLite is the most widely deployed database in the world — it’s used inside every iPhone, Android device, macOS, Firefox, Chrome, and Python installation itself. The Python sqlite3 module has been in the standard library since Python 2.5 (2006) and implements the DB-API 2.0 interface (PEP 249). Unlike MySQL or PostgreSQL, SQLite requires no server: the entire database is a single file on disk. The :memory: path creates a temporary database in RAM, perfect for testing.

⚠️ Watch out: Never build SQL queries with f-strings or string concatenation: f"INSERT INTO users VALUES ('{name}')". This is a SQL injection vulnerability — a user named O'Brien breaks the query with a syntax error, and a malicious name like '); DROP TABLE users; -- can destroy your data. Always use ? placeholders: conn.execute("INSERT INTO users VALUES (?)", (name,)). The sqlite3 module escapes the value safely.

🤔 Think about it: SQLite uses transactions — changes made by INSERT, UPDATE, or DELETE are buffered until you call conn.commit(). If your program crashes before commit, all uncommitted changes are lost. But this also means you can conn.rollback() to undo all changes since the last commit. When would you intentionally use a long-running transaction (many operations before one commit) versus committing after every single write?

Learning objectives

Use ? placeholders for parameterized queries to prevent SQL injection
Call conn.commit() after INSERT/UPDATE/DELETE to persist changes
Use fetchall() for multiple rows, fetchone() for a single row
Use :memory: as the path for an in-memory database (great for testing)
Set conn.row_factory = sqlite3.Row for dict-style column access

Key concepts

sqlite3.connect(‘:memory:’) — in-memory database
[object Object]
conn.commit() — persist transaction
cursor.fetchall() — all rows as list of tuples
conn.row_factory = sqlite3.Row — named column access

Try it

import sqlite3
def create_db():
"""Create an in-memory database with a tasks table."""
conn = sqlite3.connect(":memory:")
conn.execute(
"CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, "
"title TEXT NOT NULL, status TEXT DEFAULT ‘pending’)"
)
conn.commit()
return conn
def add_task(conn, title):
"""Add a task to the database."""
# 💡 : hint : -010 xp f-string SQL builds the query with raw user input — use "VALUES (?)", (title,) with a ? placeholder to prevent SQL injection and quote crashes
conn.execute(f"INSERT INTO tasks (title) VALUES (‘{title}’)") # BUG 1: f-string in SQL query is SQL injection vulnerability
# 💡 : hint : -010 xp SQLite buffers changes in a transaction — call conn.commit() or the INSERT is lost when the connection closes
# BUG 2: missing conn.commit() — data is buffered but never written
return True
def get_tasks(conn):
"""Return all tasks as a list of dicts."""
cursor = conn.execute("SELECT id, title, status FROM tasks")
# 💡 : hint : -010 xp fetchone() returns only the first row as a single tuple — use fetchall() to get all rows as a list
row = cursor.fetchone() # BUG 3: fetchone() returns only the first row, not all rows
if row is None:
return []
return [{"id": row[0], "title": row[1], "status": row[2]}]
def complete_task(conn, task_id):
"""Mark a task as complete."""
conn.execute(
"UPDATE tasks SET status = ‘done’ WHERE id = ?",
(task_id,)
)
conn.commit()
return True
def count_tasks(conn, status=None):
"""Count tasks, optionally filtered by status."""
if status:
cursor = conn.execute(
"SELECT COUNT() FROM tasks WHERE status = ?", (status,)
)
else:
cursor = conn.execute("SELECT COUNT() FROM tasks")
row = cursor.fetchone()
return row[0] if row else 0

Concept detail

sqlite3 — Embedded Database in Python

SQLite is a full SQL database stored in a single file (or in memory). No server needed — perfect for local apps, tests, and prototypes.

Connect and Create

import sqlite3

# In-memory database (perfect for tests — discarded when closed)
conn = sqlite3.connect(":memory:")

# File-based database
conn = sqlite3.connect("myapp.db")

conn.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
""")
conn.commit()

Parameterized Queries (ALWAYS use these)

# WRONG — SQL injection risk:
conn.execute(f"INSERT INTO users (name) VALUES ('{name}')")

# RIGHT — use ? placeholder:
conn.execute("INSERT INTO users (name) VALUES (?)", (name,))
conn.execute("SELECT * FROM users WHERE id = ?", (user_id,))

Committing Transactions

conn.execute("INSERT INTO ...")
conn.commit()  # Without this, the INSERT is NOT saved!

Fetching Results

cursor = conn.execute("SELECT id, name FROM users")
row = cursor.fetchone()     # One row or None
rows = cursor.fetchall()    # List of all rows (tuples)
# Or iterate:
for row in cursor:
    print(row[0], row[1])

Row Factory for Dict-like Access

conn.row_factory = sqlite3.Row
cursor = conn.execute("SELECT * FROM users")
for row in cursor.fetchall():
    print(row["name"])  # Access by column name

Context Manager

with sqlite3.connect(":memory:") as conn:
    conn.execute("CREATE TABLE ...")
    # conn.commit() is called automatically on exit

Solution

import sqlite3

def create_db():
    """Create an in-memory database with a tasks table."""
    conn = sqlite3.connect(":memory:")
    conn.execute(
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, "
        "title TEXT NOT NULL, status TEXT DEFAULT 'pending')"
    )
    conn.commit()
    return conn

def add_task(conn, title):
    """Add a task to the database."""
    # FIX 1: Use ? placeholder — sqlite3 escapes the value safely
    conn.execute("INSERT INTO tasks (title) VALUES (?)", (title,))
    # FIX 2: commit() persists the transaction
    conn.commit()
    return True

def get_tasks(conn):
    """Return all tasks as a list of dicts."""
    cursor = conn.execute("SELECT id, title, status FROM tasks")
    # FIX 3: fetchall() returns ALL rows
    rows = cursor.fetchall()
    return [{"id": r[0], "title": r[1], "status": r[2]} for r in rows]

def complete_task(conn, task_id):
    """Mark a task as complete."""
    conn.execute(
        "UPDATE tasks SET status = 'done' WHERE id = ?",
        (task_id,)
    )
    conn.commit()
    return True

def count_tasks(conn, status=None):
    """Count tasks, optionally filtered by status."""
    if status:
        cursor = conn.execute(
            "SELECT COUNT(*) FROM tasks WHERE status = ?", (status,)
        )
    else:
        cursor = conn.execute("SELECT COUNT(*) FROM tasks")
    row = cursor.fetchone()
    return row[0] if row else 0

Tests

def test_add_task_persists():
    """add_task must commit — task should exist after adding"""
    conn = create_db()
    add_task(conn, "Buy groceries")
    tasks = get_tasks(conn)
    assert len(tasks) == 1, "Task was not persisted — did you call conn.commit()?"
    conn.close()

def test_add_multiple_tasks_all_returned():
    """get_tasks must return ALL tasks, not just the first one"""
    conn = create_db()
    add_task(conn, "Task A")
    add_task(conn, "Task B")
    add_task(conn, "Task C")
    tasks = get_tasks(conn)
    assert len(tasks) == 3, (
        f"Expected 3 tasks but got {len(tasks)} — use fetchall(), not fetchone()"
    )
    conn.close()

def test_add_task_with_special_characters():
    """Task title with quotes must not crash — requires ? placeholder"""
    conn = create_db()
    try:
        result = add_task(conn, "Aryan's task with 'quotes' and \"double quotes\"")
        assert result is True
    except Exception as e:
        assert False, f"add_task crashed on special characters: {e}"
    conn.close()

def test_get_tasks_returns_dicts():
    conn = create_db()
    add_task(conn, "Write tests")
    tasks = get_tasks(conn)
    assert isinstance(tasks, list)
    assert isinstance(tasks[0], dict)
    assert "id" in tasks[0]
    assert "title" in tasks[0]
    assert "status" in tasks[0]
    conn.close()

def test_complete_task_changes_status():
    conn = create_db()
    add_task(conn, "Deploy app")
    tasks = get_tasks(conn)
    task_id = tasks[0]["id"]
    complete_task(conn, task_id)
    updated = get_tasks(conn)
    assert updated[0]["status"] == "done"
    conn.close()

def test_count_tasks_total():
    conn = create_db()
    add_task(conn, "A")
    add_task(conn, "B")
    add_task(conn, "C")
    assert count_tasks(conn) == 3
    conn.close()

def test_count_tasks_by_status():
    conn = create_db()
    add_task(conn, "Task 1")
    add_task(conn, "Task 2")
    task_id = get_tasks(conn)[0]["id"]
    complete_task(conn, task_id)
    assert count_tasks(conn, "done") == 1
    assert count_tasks(conn, "pending") == 1
    conn.close()

def test_empty_db_returns_empty_list():
    conn = create_db()
    assert get_tasks(conn) == []
    conn.close()