Skip to main content

๐Ÿ—„๏ธ Lesson 11: Database Basics with SQLite

Learn to store, query, and manage structured data using SQL and Python's built-in sqlite3 module. You'll master CRUD operations, parameterized queries, table relationships, and build a data-backed contact book application.

๐ŸŽฏ Learning Objectives

By the end of this lesson, you will be able to:

  • Explain what a relational database is and when to use one
  • Connect to a SQLite database with Python's sqlite3 module
  • Create tables with appropriate column types and constraints
  • Perform CRUD operations โ€” INSERT, SELECT, UPDATE, DELETE
  • Use parameterized queries to prevent SQL injection
  • Filter, sort, and aggregate data with WHERE, ORDER BY, and aggregate functions
  • Create relationships between tables with foreign keys and JOIN
  • Use the with statement for safe connection management

Estimated Time: 60 minutes

Prerequisites: Dictionaries, tuples, context managers (Lesson 4), error handling (Lesson 5)

In This Lesson

๐Ÿ’พ Why Databases?

So far you've stored data in variables, lists, dicts, and files. These work fine for small scripts, but they hit walls fast:

Approach Limitation
Variables / dicts Gone when the program exits
Text / CSV files Hard to search, filter, or update specific records
JSON files Must load entire file into memory; no built-in querying

A database solves all of these problems. It stores data on disk (persistent), supports powerful querying (find exactly what you need), handles concurrent access, and enforces data integrity with constraints.

Relational Databases

A relational database organizes data into tables (like spreadsheets) with rows (records) and columns (fields). Tables can reference each other through relationships.

๐Ÿ“‹ students id name grade email 1 Ada Lovelace 95 ada@ex.com 2 Alan Turing 91 alan@ex.com 3 Grace Hopper 88 grace@ex.com โ† Row โ†‘ Columns

You interact with relational databases using SQL (Structured Query Language) โ€” a declarative language designed specifically for data. You tell the database what data you want, not how to find it.

๐Ÿชถ SQLite & the sqlite3 Module

SQLite is a lightweight, file-based database engine. Unlike MySQL or PostgreSQL, it doesn't require a separate server โ€” the entire database lives in a single file on disk. It's built into Python's standard library, so there's nothing to install.

โœ… Why SQLite Is Perfect for Learning

  • Zero setup โ€” no server, no configuration, no accounts
  • Built into Python โ€” import sqlite3 and go
  • Single file โ€” your database is just a .db file you can copy, share, or delete
  • Real SQL โ€” the SQL you learn here transfers to PostgreSQL, MySQL, etc.
  • Production-worthy โ€” used in Android, iOS, browsers, and countless applications

Connecting to a Database

import sqlite3

# Connect to a database file (creates it if it doesn't exist)
conn = sqlite3.connect("my_database.db")

# Create a cursor โ€” this object executes SQL commands
cursor = conn.cursor()

# ... do work with the database ...

# Always close when done
conn.close()

The Better Way: Context Manager

Use the with statement so the connection is properly handled even if errors occur. The connection object auto-commits on success and rolls back on exception:

import sqlite3

with sqlite3.connect("my_database.db") as conn:
    cursor = conn.cursor()
    # ... work with database ...
    # Auto-commits if no exception, auto-rolls-back on exception
graph LR A["๐Ÿ Python"] -->|"sqlite3.connect()"| B["๐Ÿ“‚ .db File"] B -->|"conn.cursor()"| C["๐Ÿ“ Cursor"] C -->|"cursor.execute(sql)"| D["โš™๏ธ SQL Engine"] D -->|"results"| C C -->|"conn.commit()"| B style A fill:#eff6ff,stroke:#3b82f6,color:#1e293b style B fill:#fefce8,stroke:#f59e0b,color:#1e293b style C fill:#f0fdf4,stroke:#22c55e,color:#1e293b style D fill:#f8fafc,stroke:#6366f1,color:#1e293b

๐Ÿง  In-Memory Databases

Pass ":memory:" instead of a filename to create a temporary database that lives entirely in RAM. Great for testing and throwaway work:

conn = sqlite3.connect(":memory:")

๐Ÿ—๏ธ Creating Tables

The CREATE TABLE statement defines the structure of your data โ€” column names, data types, and constraints:

import sqlite3

with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS students (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE,
            grade REAL DEFAULT 0.0,
            enrolled_date TEXT DEFAULT CURRENT_DATE
        )
    """)

SQLite Data Types

SQLite uses a simple type system โ€” every value is one of five storage classes:

SQLite Type Python Type Use For
INTEGER int Whole numbers, IDs, counts
REAL float Decimals, grades, prices
TEXT str Names, emails, dates, descriptions
BLOB bytes Binary data (images, files)
NULL None Missing or unknown values

Column Constraints

Constraints enforce rules on your data to keep it valid:

Constraint Meaning
PRIMARY KEY Unique identifier for each row
AUTOINCREMENT Auto-assigns the next integer
NOT NULL Column cannot be empty
UNIQUE No two rows can share this value
DEFAULT value Value used when none is provided
CHECK(expr) Custom validation rule
# Example with CHECK constraint
cursor.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        price REAL NOT NULL CHECK(price >= 0),
        quantity INTEGER NOT NULL DEFAULT 0 CHECK(quantity >= 0)
    )
""")

โš ๏ธ IF NOT EXISTS

Always use CREATE TABLE IF NOT EXISTS โ€” otherwise you'll get an error if the table already exists when your script runs a second time.

โž• Inserting Data

Use INSERT INTO to add rows. Always use placeholders (?) instead of f-strings โ€” we'll explain why in the SQL Injection section.

Insert a Single Row

import sqlite3

with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()

    cursor.execute("""
        INSERT INTO students (name, email, grade)
        VALUES (?, ?, ?)
    """, ("Ada Lovelace", "ada@example.com", 95.5))

    print(f"Inserted row with id: {cursor.lastrowid}")

Insert Multiple Rows

Use executemany() to insert a batch of rows efficiently:

students_data = [
    ("Alan Turing", "alan@example.com", 91.0),
    ("Grace Hopper", "grace@example.com", 88.5),
    ("Linus Torvalds", "linus@example.com", 92.0),
    ("Guido van Rossum", "guido@example.com", 97.0),
]

with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()

    cursor.executemany("""
        INSERT INTO students (name, email, grade)
        VALUES (?, ?, ?)
    """, students_data)

    print(f"Inserted {cursor.rowcount} rows")

๐Ÿ’ก Commit Behavior

When using the with statement, SQLite auto-commits when the block exits without an error. If you're not using with, you must call conn.commit() explicitly โ€” otherwise your changes won't be saved!

๐Ÿ” Querying Data

The SELECT statement retrieves data from a table. After executing a query, you use the cursor's fetch methods to read results:

Fetch Methods

Method Returns Use When
fetchone() One row (tuple) or None You expect exactly 0 or 1 result
fetchall() List of all rows (tuples) You want every result in memory
fetchmany(n) List of up to n rows Large result sets โ€” process in batches
Iterate cursor One row at a time Memory-efficient for large results
import sqlite3

with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()

    # Get all students
    cursor.execute("SELECT * FROM students")
    all_students = cursor.fetchall()
    for row in all_students:
        print(row)
    # (1, 'Ada Lovelace', 'ada@example.com', 95.5, '2026-04-13')
    # (2, 'Alan Turing', 'alan@example.com', 91.0, '2026-04-13')
    # ...

    # Select specific columns
    cursor.execute("SELECT name, grade FROM students")
    for name, grade in cursor.fetchall():
        print(f"{name}: {grade}")

    # Get one student
    cursor.execute("SELECT * FROM students WHERE id = ?", (1,))
    student = cursor.fetchone()
    print(student)  # (1, 'Ada Lovelace', 'ada@example.com', 95.5, '...')

Iterating Directly Over the Cursor

For large result sets, iterate directly over the cursor instead of loading everything with fetchall():

cursor.execute("SELECT name, grade FROM students")
for name, grade in cursor:
    print(f"{name}: {grade}")

Getting Column Names

cursor.execute("SELECT * FROM students")
column_names = [desc[0] for desc in cursor.description]
print(column_names)  # ['id', 'name', 'email', 'grade', 'enrolled_date']

โœ… Row Factory: Dict-Like Rows

Tired of accessing columns by index? Set row_factory to get dict-like rows:

conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM students WHERE id = ?", (1,))
student = cursor.fetchone()

# Access by name โ€” much cleaner!
print(student["name"])    # "Ada Lovelace"
print(student["grade"])   # 95.5
print(dict(student))      # Convert to a real dict

๐ŸŽฏ Filtering & Sorting

WHERE โ€” Filtering Rows

The WHERE clause filters which rows are returned:

# Students with grade above 90
cursor.execute("SELECT name, grade FROM students WHERE grade > ?", (90,))

# Exact match
cursor.execute("SELECT * FROM students WHERE name = ?", ("Ada Lovelace",))

# Pattern matching with LIKE
cursor.execute("SELECT * FROM students WHERE name LIKE ?", ("%Turing%",))

# Multiple conditions with AND / OR
cursor.execute("""
    SELECT name, grade FROM students
    WHERE grade >= ? AND grade <= ?
""", (85, 95))

# IN โ€” match any value in a list
cursor.execute("""
    SELECT * FROM students WHERE name IN (?, ?, ?)
""", ("Ada Lovelace", "Alan Turing", "Grace Hopper"))

# NULL checks
cursor.execute("SELECT * FROM students WHERE email IS NOT NULL")

ORDER BY โ€” Sorting Results

# Ascending (default)
cursor.execute("SELECT name, grade FROM students ORDER BY grade")

# Descending โ€” highest first
cursor.execute("SELECT name, grade FROM students ORDER BY grade DESC")

# Multiple sort keys
cursor.execute("""
    SELECT name, grade FROM students
    ORDER BY grade DESC, name ASC
""")

LIMIT & OFFSET โ€” Pagination

# Top 3 students
cursor.execute("""
    SELECT name, grade FROM students
    ORDER BY grade DESC
    LIMIT 3
""")

# Skip first 3, get next 3 (page 2)
cursor.execute("""
    SELECT name, grade FROM students
    ORDER BY grade DESC
    LIMIT 3 OFFSET 3
""")
graph LR S["SELECT"] --> F["FROM"] --> W["WHERE"] --> O["ORDER BY"] --> L["LIMIT"] style S fill:#eff6ff,stroke:#3b82f6,color:#1e293b style F fill:#f0fdf4,stroke:#22c55e,color:#1e293b style W fill:#fefce8,stroke:#f59e0b,color:#1e293b style O fill:#faf5ff,stroke:#a855f7,color:#1e293b style L fill:#fff1f2,stroke:#ef4444,color:#1e293b

๐Ÿ’ก SQL Clause Order

SQL clauses must appear in this order: SELECT โ†’ FROM โ†’ WHERE โ†’ GROUP BY โ†’ HAVING โ†’ ORDER BY โ†’ LIMIT. You can skip any clause, but you can't reorder them.

โœ๏ธ Updating & Deleting

UPDATE โ€” Modify Existing Rows

# Update one student's grade
cursor.execute("""
    UPDATE students SET grade = ? WHERE id = ?
""", (98.0, 1))

# Update multiple columns
cursor.execute("""
    UPDATE students
    SET email = ?, grade = ?
    WHERE name = ?
""", ("new_ada@example.com", 99.0, "Ada Lovelace"))

print(f"Rows updated: {cursor.rowcount}")

DELETE โ€” Remove Rows

# Delete a specific student
cursor.execute("DELETE FROM students WHERE id = ?", (3,))

# Delete students with low grades
cursor.execute("DELETE FROM students WHERE grade < ?", (60,))

print(f"Rows deleted: {cursor.rowcount}")

๐Ÿšจ Always Use WHERE with UPDATE and DELETE!

Without a WHERE clause, UPDATE changes every row and DELETE removes every row in the table. This is the most common SQL mistake โ€” always double-check your WHERE condition.

-- โŒ DANGER: This deletes ALL students!
DELETE FROM students;

-- โœ… SAFE: Only deletes one specific student
DELETE FROM students WHERE id = 3;

๐Ÿ›ก๏ธ Parameterized Queries & SQL Injection

SQL injection is one of the most common and dangerous security vulnerabilities. It happens when user input is inserted directly into SQL strings:

# โŒ NEVER DO THIS โ€” vulnerable to SQL injection!
name = input("Enter student name: ")
cursor.execute(f"SELECT * FROM students WHERE name = '{name}'")

If the user enters '; DROP TABLE students; -- instead of a name, the query becomes:

SELECT * FROM students WHERE name = ''; DROP TABLE students; --'

That deletes your entire table. This is known as the "Bobby Tables" attack.

โŒ String Formatting f"...WHERE name = '{name}'" User input โ†’ SQL code! โœ… Parameterized Query "...WHERE name = ?", (name,) User input โ†’ safe data only ๐Ÿ”‘ Rule: ALWAYS use ? placeholders for user data. Never use f-strings, .format(), or % with SQL.

The Fix: Parameterized Queries

Parameterized queries treat user input as data, never as SQL code. The database engine handles escaping and quoting automatically:

# โœ… SAFE โ€” parameterized query
name = input("Enter student name: ")
cursor.execute("SELECT * FROM students WHERE name = ?", (name,))

# โœ… Multiple parameters
cursor.execute("""
    SELECT * FROM students
    WHERE grade >= ? AND grade <= ?
""", (min_grade, max_grade))

# โœ… Named parameters (alternative style)
cursor.execute("""
    SELECT * FROM students
    WHERE name = :name AND grade >= :min_grade
""", {"name": "Ada Lovelace", "min_grade": 90})

๐Ÿง  Why Not Just Escape Special Characters?

Manual escaping is error-prone and always one edge case away from failure. Parameterized queries separate code from data at the protocol level โ€” the database knows the difference before it ever sees the values. Use ? placeholders. Every time. No exceptions.

๐Ÿ“Š Aggregate Functions

Aggregate functions compute a single value from a set of rows:

# Count all students
cursor.execute("SELECT COUNT(*) FROM students")
total = cursor.fetchone()[0]
print(f"Total students: {total}")

# Average grade
cursor.execute("SELECT AVG(grade) FROM students")
avg = cursor.fetchone()[0]
print(f"Average grade: {avg:.1f}")

# Highest and lowest grades
cursor.execute("SELECT MAX(grade), MIN(grade) FROM students")
high, low = cursor.fetchone()
print(f"Highest: {high}, Lowest: {low}")

# Sum of all grades (maybe for a total points system)
cursor.execute("SELECT SUM(grade) FROM students")
total_points = cursor.fetchone()[0]
print(f"Total points: {total_points}")

GROUP BY โ€” Aggregate Per Group

Group rows by a column and compute aggregates for each group:

# Suppose students have a 'year' column
cursor.execute("""
    SELECT enrolled_date, COUNT(*), AVG(grade)
    FROM students
    GROUP BY enrolled_date
""")
for date, count, avg_grade in cursor:
    print(f"  {date}: {count} students, avg grade {avg_grade:.1f}")

# HAVING โ€” filter groups (like WHERE but for aggregates)
cursor.execute("""
    SELECT enrolled_date, AVG(grade) as avg_g
    FROM students
    GROUP BY enrolled_date
    HAVING avg_g > 90
""")
Function Returns
COUNT(*) Number of rows
COUNT(col) Number of non-NULL values in column
AVG(col) Average (mean) value
SUM(col) Sum of all values
MAX(col) Largest value
MIN(col) Smallest value

๐Ÿ”— Table Relationships & JOINs

Real applications need multiple related tables. A foreign key links a column in one table to the primary key of another:

import sqlite3

with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()

    # Enable foreign key enforcement (off by default in SQLite!)
    cursor.execute("PRAGMA foreign_keys = ON")

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS courses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            instructor TEXT
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS enrollments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            student_id INTEGER NOT NULL,
            course_id INTEGER NOT NULL,
            grade REAL,
            FOREIGN KEY (student_id) REFERENCES students(id),
            FOREIGN KEY (course_id) REFERENCES courses(id)
        )
    """)
erDiagram students { INTEGER id PK TEXT name TEXT email REAL grade } courses { INTEGER id PK TEXT name TEXT instructor } enrollments { INTEGER id PK INTEGER student_id FK INTEGER course_id FK REAL grade } students ||--o{ enrollments : "has" courses ||--o{ enrollments : "has"

JOIN โ€” Combine Tables

A JOIN combines rows from two tables based on a related column. The most common type is INNER JOIN, which returns only rows that have matches in both tables:

# Insert sample data
cursor.executemany("INSERT INTO courses (name, instructor) VALUES (?, ?)", [
    ("Python 101", "Dr. Van Rossum"),
    ("Algorithms", "Prof. Knuth"),
])

cursor.executemany("""
    INSERT INTO enrollments (student_id, course_id, grade) VALUES (?, ?, ?)
""", [
    (1, 1, 95.0),  # Ada in Python 101
    (1, 2, 89.0),  # Ada in Algorithms
    (2, 1, 91.0),  # Alan in Python 101
])

# INNER JOIN โ€” students with their courses
cursor.execute("""
    SELECT s.name, c.name AS course, e.grade
    FROM enrollments e
    INNER JOIN students s ON e.student_id = s.id
    INNER JOIN courses c ON e.course_id = c.id
    ORDER BY s.name, c.name
""")
for student, course, grade in cursor:
    print(f"  {student} โ€” {course}: {grade}")

Output:

  Ada Lovelace โ€” Algorithms: 89.0
  Ada Lovelace โ€” Python 101: 95.0
  Alan Turing โ€” Python 101: 91.0

Types of JOINs

Join Type Returns
INNER JOIN Only rows with matches in both tables
LEFT JOIN All rows from left table, matched rows from right (NULL if no match)
CROSS JOIN Every combination of rows from both tables (Cartesian product)
# LEFT JOIN โ€” show ALL students, even those not enrolled in anything
cursor.execute("""
    SELECT s.name, c.name AS course
    FROM students s
    LEFT JOIN enrollments e ON s.id = e.student_id
    LEFT JOIN courses c ON e.course_id = c.id
""")
for student, course in cursor:
    print(f"  {student} โ€” {course or '(no courses)'}")

๐Ÿ’ก PRAGMA foreign_keys = ON

SQLite disables foreign key enforcement by default for backward compatibility. Always run PRAGMA foreign_keys = ON after connecting if you want the database to actually enforce your FOREIGN KEY constraints. Without it, you can insert invalid references without error.

๐Ÿ—๏ธ Mini-Project: Contact Book

Let's build a complete CLI contact book that stores data in SQLite. This brings together everything from this lesson: table creation, CRUD, parameterized queries, and error handling.

"""
contact_book.py โ€” A CLI contact book backed by SQLite.
"""
import sqlite3


DB_FILE = "contacts.db"


def get_connection():
    """Create and return a database connection with row_factory."""
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def init_db():
    """Create the contacts table if it doesn't exist."""
    with get_connection() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS contacts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                phone TEXT,
                email TEXT UNIQUE,
                category TEXT DEFAULT 'general',
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
            )
        """)


def add_contact(name, phone=None, email=None, category="general"):
    """Add a new contact. Returns the new contact's id."""
    with get_connection() as conn:
        try:
            cursor = conn.execute("""
                INSERT INTO contacts (name, phone, email, category)
                VALUES (?, ?, ?, ?)
            """, (name, phone, email, category))
            print(f"โœ… Added: {name} (id={cursor.lastrowid})")
            return cursor.lastrowid
        except sqlite3.IntegrityError as e:
            print(f"โŒ Could not add contact: {e}")
            return None


def list_contacts(category=None):
    """List all contacts, optionally filtered by category."""
    with get_connection() as conn:
        if category:
            cursor = conn.execute(
                "SELECT * FROM contacts WHERE category = ? ORDER BY name",
                (category,)
            )
        else:
            cursor = conn.execute("SELECT * FROM contacts ORDER BY name")

        contacts = cursor.fetchall()
        if not contacts:
            print("๐Ÿ“ญ No contacts found.")
            return

        print(f"\n{'ID':<5} {'Name':<25} {'Phone':<15} {'Email':<30} {'Category':<10}")
        print("-" * 85)
        for c in contacts:
            print(f"{c['id']:<5} {c['name']:<25} {c['phone'] or 'โ€”':<15} "
                  f"{c['email'] or 'โ€”':<30} {c['category']:<10}")
        print(f"\n  Total: {len(contacts)} contact(s)")


def search_contacts(query):
    """Search contacts by name, phone, or email."""
    with get_connection() as conn:
        cursor = conn.execute("""
            SELECT * FROM contacts
            WHERE name LIKE ? OR phone LIKE ? OR email LIKE ?
            ORDER BY name
        """, (f"%{query}%", f"%{query}%", f"%{query}%"))

        results = cursor.fetchall()
        if not results:
            print(f"๐Ÿ” No contacts matching '{query}'.")
            return

        print(f"\n๐Ÿ” Found {len(results)} result(s) for '{query}':")
        for c in results:
            print(f"  [{c['id']}] {c['name']} โ€” {c['phone'] or 'no phone'} โ€” {c['email'] or 'no email'}")


def update_contact(contact_id, **fields):
    """Update a contact's fields by id."""
    if not fields:
        print("Nothing to update.")
        return

    set_clause = ", ".join(f"{key} = ?" for key in fields)
    values = list(fields.values()) + [contact_id]

    with get_connection() as conn:
        cursor = conn.execute(
            f"UPDATE contacts SET {set_clause} WHERE id = ?", values
        )
        if cursor.rowcount:
            print(f"โœ… Updated contact {contact_id}.")
        else:
            print(f"โŒ No contact with id {contact_id}.")


def delete_contact(contact_id):
    """Delete a contact by id."""
    with get_connection() as conn:
        cursor = conn.execute(
            "DELETE FROM contacts WHERE id = ?", (contact_id,)
        )
        if cursor.rowcount:
            print(f"๐Ÿ—‘๏ธ Deleted contact {contact_id}.")
        else:
            print(f"โŒ No contact with id {contact_id}.")


def show_stats():
    """Show database statistics."""
    with get_connection() as conn:
        total = conn.execute("SELECT COUNT(*) FROM contacts").fetchone()[0]
        categories = conn.execute("""
            SELECT category, COUNT(*) as cnt
            FROM contacts GROUP BY category ORDER BY cnt DESC
        """).fetchall()

    print(f"\n๐Ÿ“Š Contact Book Stats")
    print(f"  Total contacts: {total}")
    if categories:
        print(f"  By category:")
        for row in categories:
            print(f"    {row['category']}: {row['cnt']}")


def main():
    """Interactive contact book CLI."""
    init_db()
    print("๐Ÿ“’ Contact Book")
    print("Commands: add, list, search, update, delete, stats, quit\n")

    while True:
        command = input(">>> ").strip().lower()

        if command == "quit":
            print("Goodbye! ๐Ÿ‘‹")
            break

        elif command == "add":
            name = input("  Name: ").strip()
            if not name:
                print("  Name is required.")
                continue
            phone = input("  Phone (optional): ").strip() or None
            email = input("  Email (optional): ").strip() or None
            category = input("  Category [general]: ").strip() or "general"
            add_contact(name, phone, email, category)

        elif command == "list":
            cat = input("  Category filter (or Enter for all): ").strip() or None
            list_contacts(cat)

        elif command == "search":
            query = input("  Search: ").strip()
            if query:
                search_contacts(query)

        elif command == "update":
            try:
                cid = int(input("  Contact ID: "))
            except ValueError:
                print("  Invalid ID.")
                continue
            print("  Enter new values (or Enter to skip):")
            fields = {}
            for field in ("name", "phone", "email", "category"):
                val = input(f"    {field}: ").strip()
                if val:
                    fields[field] = val
            update_contact(cid, **fields)

        elif command == "delete":
            try:
                cid = int(input("  Contact ID: "))
            except ValueError:
                print("  Invalid ID.")
                continue
            confirm = input(f"  Delete contact {cid}? (y/n): ").strip().lower()
            if confirm == "y":
                delete_contact(cid)

        elif command == "stats":
            show_stats()

        else:
            print("Unknown command. Try: add, list, search, update, delete, stats, quit")


if __name__ == "__main__":
    main()

Sample Session:

๐Ÿ“’ Contact Book
Commands: add, list, search, update, delete, stats, quit

>>> add
  Name: Maria Santos
  Phone (optional): 555-0101
  Email (optional): maria@example.com
  Category [general]: friend
โœ… Added: Maria Santos (id=1)

>>> add
  Name: Dr. Rivera
  Phone (optional): 555-0202
  Email (optional): rivera@clinic.com
  Category [general]: medical
โœ… Added: Dr. Rivera (id=2)

>>> list

ID    Name                      Phone           Email                          Category
-------------------------------------------------------------------------------------
2     Dr. Rivera                555-0202        rivera@clinic.com              medical
1     Maria Santos              555-0101        maria@example.com              friend

  Total: 2 contact(s)

>>> search
  Search: maria
๐Ÿ” Found 1 result(s) for 'maria':
  [1] Maria Santos โ€” 555-0101 โ€” maria@example.com

>>> stats
๐Ÿ“Š Contact Book Stats
  Total contacts: 2
  By category:
    friend: 1
    medical: 1

>>> quit
Goodbye! ๐Ÿ‘‹

๐Ÿ’ช Hands-on Exercises

Exercise 1: Inventory Tracker

Create a simple inventory management system. Write functions to:

  1. Create a products table with: id (auto PK), name (text, not null, unique), price (real, โ‰ฅ 0), quantity (integer, โ‰ฅ 0)
  2. add_product(name, price, quantity) โ€” inserts a product
  3. restock(product_id, amount) โ€” increases quantity by amount
  4. sell(product_id, amount) โ€” decreases quantity by amount (but not below 0)
  5. low_stock_report(threshold=5) โ€” prints products with quantity โ‰ค threshold
๐Ÿ’ก Hint

For sell(), first check the current quantity with a SELECT, then only UPDATE if there's enough stock. Use CHECK(quantity >= 0) on the table as a safety net. For restock(), use UPDATE products SET quantity = quantity + ? WHERE id = ? โ€” you can do arithmetic directly in SQL!

โœ… Solution
import sqlite3

DB = "inventory.db"


def get_conn():
    conn = sqlite3.connect(DB)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    with get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS products (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE,
                price REAL NOT NULL CHECK(price >= 0),
                quantity INTEGER NOT NULL DEFAULT 0 CHECK(quantity >= 0)
            )
        """)


def add_product(name, price, quantity=0):
    with get_conn() as conn:
        try:
            cursor = conn.execute(
                "INSERT INTO products (name, price, quantity) VALUES (?, ?, ?)",
                (name, price, quantity)
            )
            print(f"Added '{name}' (id={cursor.lastrowid})")
        except sqlite3.IntegrityError as e:
            print(f"Error: {e}")


def restock(product_id, amount):
    if amount <= 0:
        print("Amount must be positive.")
        return
    with get_conn() as conn:
        cursor = conn.execute(
            "UPDATE products SET quantity = quantity + ? WHERE id = ?",
            (amount, product_id)
        )
        if cursor.rowcount:
            print(f"Restocked product {product_id} by {amount}.")
        else:
            print(f"Product {product_id} not found.")


def sell(product_id, amount):
    if amount <= 0:
        print("Amount must be positive.")
        return
    with get_conn() as conn:
        row = conn.execute(
            "SELECT name, quantity FROM products WHERE id = ?",
            (product_id,)
        ).fetchone()
        if not row:
            print(f"Product {product_id} not found.")
            return
        if row["quantity"] < amount:
            print(f"Not enough stock for '{row['name']}' "
                  f"(have {row['quantity']}, need {amount}).")
            return
        conn.execute(
            "UPDATE products SET quantity = quantity - ? WHERE id = ?",
            (amount, product_id)
        )
        print(f"Sold {amount}x '{row['name']}'. "
              f"Remaining: {row['quantity'] - amount}")


def low_stock_report(threshold=5):
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM products WHERE quantity <= ? ORDER BY quantity",
            (threshold,)
        ).fetchall()
    if not rows:
        print("All products are well-stocked!")
        return
    print(f"\nโš ๏ธ Low Stock (โ‰ค {threshold}):")
    for r in rows:
        print(f"  [{r['id']}] {r['name']}: {r['quantity']} left (${r['price']:.2f})")


if __name__ == "__main__":
    init_db()
    add_product("Widget", 9.99, 10)
    add_product("Gadget", 24.95, 3)
    add_product("Doohickey", 4.50, 1)
    sell(1, 7)
    restock(3, 20)
    low_stock_report()

Exercise 2: Reading Log with JOIN

Create a reading log database with two tables:

  1. books โ€” id (auto PK), title (not null), author (not null), pages (integer)
  2. reading_log โ€” id (auto PK), book_id (foreign key โ†’ books.id), date_read (text, default CURRENT_DATE), pages_read (integer), notes (text)

Write functions to:

  • add_book(title, author, pages)
  • log_reading(book_id, pages_read, notes="")
  • reading_summary() โ€” use a JOIN to show each book's title, total pages read, and number of reading sessions
๐Ÿ’ก Hint

For reading_summary(), use a LEFT JOIN from books to reading_log with GROUP BY, SUM(pages_read), and COUNT(reading_log.id). The LEFT JOIN ensures books with no reading entries still show up.

โœ… Solution
import sqlite3

DB = "reading_log.db"


def get_conn():
    conn = sqlite3.connect(DB)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def init_db():
    with get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS books (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                author TEXT NOT NULL,
                pages INTEGER
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS reading_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                book_id INTEGER NOT NULL,
                date_read TEXT DEFAULT CURRENT_DATE,
                pages_read INTEGER NOT NULL,
                notes TEXT,
                FOREIGN KEY (book_id) REFERENCES books(id)
            )
        """)


def add_book(title, author, pages=None):
    with get_conn() as conn:
        cursor = conn.execute(
            "INSERT INTO books (title, author, pages) VALUES (?, ?, ?)",
            (title, author, pages)
        )
        print(f"Added book: '{title}' by {author} (id={cursor.lastrowid})")
        return cursor.lastrowid


def log_reading(book_id, pages_read, notes=""):
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO reading_log (book_id, pages_read, notes) VALUES (?, ?, ?)",
            (book_id, pages_read, notes)
        )
        print(f"Logged {pages_read} pages for book {book_id}.")


def reading_summary():
    with get_conn() as conn:
        rows = conn.execute("""
            SELECT
                b.title,
                b.author,
                b.pages AS total_pages,
                COUNT(rl.id) AS sessions,
                COALESCE(SUM(rl.pages_read), 0) AS pages_read
            FROM books b
            LEFT JOIN reading_log rl ON b.id = rl.book_id
            GROUP BY b.id
            ORDER BY pages_read DESC
        """).fetchall()

    print(f"\n๐Ÿ“š Reading Summary")
    print(f"{'Title':<30} {'Author':<20} {'Read':<12} {'Sessions':<10}")
    print("-" * 72)
    for r in rows:
        total = f"{r['pages_read']}/{r['total_pages']}p" if r['total_pages'] else f"{r['pages_read']}p"
        print(f"{r['title']:<30} {r['author']:<20} {total:<12} {r['sessions']:<10}")


if __name__ == "__main__":
    init_db()
    b1 = add_book("Dune", "Frank Herbert", 412)
    b2 = add_book("Neuromancer", "William Gibson", 271)
    log_reading(b1, 50, "Just started โ€” worldbuilding is incredible")
    log_reading(b1, 80, "The spice must flow")
    log_reading(b2, 30, "Cyberpunk classic")
    reading_summary()

๐Ÿ“ Summary

You now know how to store and manage persistent, structured data with SQLite โ€” a skill that powers everything from mobile apps to data science pipelines. Here's what we covered:

  • Relational databases organize data into tables with rows and columns, queried with SQL
  • SQLite is file-based, built into Python, and requires zero setup
  • CREATE TABLE defines your schema โ€” columns, types (INTEGER, TEXT, REAL), and constraints (PRIMARY KEY, NOT NULL, UNIQUE, CHECK)
  • INSERT adds data; use executemany() for batches
  • SELECT queries data with fetchone(), fetchall(), or cursor iteration
  • WHERE filters, ORDER BY sorts, LIMIT paginates
  • UPDATE and DELETE modify data โ€” always include a WHERE clause!
  • Parameterized queries (? placeholders) prevent SQL injection โ€” never use f-strings with SQL
  • Aggregate functions (COUNT, AVG, SUM, MAX, MIN) summarize data; GROUP BY groups them
  • Foreign keys create relationships; JOIN combines related tables
  • sqlite3.Row gives dict-like access to columns by name

๐Ÿ”‘ Quick Reference

Task SQL / Python
Connect sqlite3.connect("file.db")
Create table CREATE TABLE IF NOT EXISTS t (...)
Insert INSERT INTO t (col) VALUES (?)
Query SELECT col FROM t WHERE ...
Update UPDATE t SET col = ? WHERE id = ?
Delete DELETE FROM t WHERE id = ?
Join SELECT ... FROM a JOIN b ON a.id = b.a_id
Aggregate SELECT COUNT(*), AVG(col) FROM t GROUP BY ...
Dict-like rows conn.row_factory = sqlite3.Row
Foreign keys PRAGMA foreign_keys = ON

๐Ÿ“š Additional Resources

๐Ÿš€ What's Next?

In the final lesson, you'll combine everything from this course into a Capstone Project โ€” a CLI Task Manager with OOP, SQLite, API integration, proper project structure, and a test suite. It's your chance to prove you've leveled up.

๐ŸŽฏ Almost There!

You've now mastered both APIs and databases โ€” the two pillars of real-world data handling. One more lesson and you'll have completed the entire Intermediate Python course!

๐Ÿงช Knowledge Check

Test your understanding with these quick questions:

Question 1

Why is this code dangerous?
cursor.execute(f"DELETE FROM users WHERE id = {user_input}")

Question 2

What does cursor.fetchone() return when there are no matching rows?

Question 3

Which SQL clause would you use to get the average grade for each course?