๐๏ธ 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
sqlite3module - 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
withstatement 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.
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 sqlite3and go - Single file โ your database is just a
.dbfile 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
๐ง 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
""")
๐ก 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.
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)
)
""")
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:
- Create a
productstable with:id(auto PK),name(text, not null, unique),price(real, โฅ 0),quantity(integer, โฅ 0) add_product(name, price, quantity)โ inserts a productrestock(product_id, amount)โ increases quantity by amountsell(product_id, amount)โ decreases quantity by amount (but not below 0)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:
booksโid(auto PK),title(not null),author(not null),pages(integer)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 TABLEdefines your schema โ columns, types (INTEGER,TEXT,REAL), and constraints (PRIMARY KEY,NOT NULL,UNIQUE,CHECK)INSERTadds data; useexecutemany()for batchesSELECTqueries data withfetchone(),fetchall(), or cursor iterationWHEREfilters,ORDER BYsorts,LIMITpaginatesUPDATEandDELETEmodify data โ always include aWHEREclause!- Parameterized queries (
?placeholders) prevent SQL injection โ never use f-strings with SQL - Aggregate functions (
COUNT,AVG,SUM,MAX,MIN) summarize data;GROUP BYgroups them - Foreign keys create relationships;
JOINcombines related tables sqlite3.Rowgives 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
- Python Docs โ sqlite3 Module
- SQLite SQL Language Reference
- SQLite Tutorial โ Comprehensive SQL Guide
- SQLZOO โ Interactive SQL Practice
- Bobby Tables โ SQL Injection Guide
๐ 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?