Databases with SQLite 🗄️

When a file is not enough

Reach for a database when you need any of:

• to find things by criteria without loading everything into memory;
• relationships between kinds of thing (orders belong to customers);
• more data than fits comfortably in RAM;
• several things writing at once without corrupting each other;
• a guarantee that a half-finished update never survives a crash.

SQLite gives you all of that in a file, with no server to install or administer. It is in your phone, your browser, and on most aircraft. The official docs claim it is the most widely deployed database engine in the world, and that is probably true.

Creating and inserting

import sqlite3

conn = sqlite3.connect("ship.db")
conn.execute("""
    CREATE TABLE IF NOT EXISTS crew (
        id       INTEGER PRIMARY KEY,
        name     TEXT NOT NULL,
        role     TEXT NOT NULL DEFAULT 'deckhand',
        pay      INTEGER NOT NULL CHECK (pay >= 0),
        joined   TEXT NOT NULL
    )
""")

conn.execute(
    "INSERT INTO crew (name, role, pay, joined) VALUES (?, ?, ?, ?)",
    ("Guybrush", "captain", 100, "1990-10-15"),
)
conn.executemany(
    "INSERT INTO crew (name, role, pay, joined) VALUES (?, ?, ?, ?)",
    [
        ("Elaine", "governor", 250, "1990-10-15"),
        ("Otis", "lookout", 40, "1991-01-03"),
        ("Meathook", "lookout", 45, "1991-06-20"),
    ],
)
conn.commit()

print(conn.execute("SELECT COUNT(*) FROM crew").fetchone()[0], "crew")
conn.close()

4 crew

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (name TEXT)")
conn.execute("INSERT INTO users VALUES ('Guybrush')")
conn.execute("INSERT INTO users VALUES ('Elaine')")

attack = "' OR '1'='1"

# The safe way: the input is data, so it simply matches nothing
safe = conn.execute("SELECT * FROM users WHERE name = ?", (attack,)).fetchall()
print("parameterised:", safe)

# The unsafe way, shown once so you recognise it
unsafe = conn.execute(f"SELECT * FROM users WHERE name = '{attack}'").fetchall()
print("f-string:     ", unsafe)

parameterised: []
f-string:      [('Guybrush',), ('Elaine',)]

The second query returned every row, because the input closed the quote and added its own condition. That is SQL injection, demonstrated in four lines.

Querying

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE crew (id INTEGER PRIMARY KEY, name TEXT, role TEXT, pay INTEGER)")
conn.executemany("INSERT INTO crew (name, role, pay) VALUES (?, ?, ?)", [
    ("Guybrush", "captain", 100),
    ("Elaine", "governor", 250),
    ("Otis", "lookout", 40),
    ("Meathook", "lookout", 45),
])

for row in conn.execute("SELECT name, pay FROM crew WHERE pay > ? ORDER BY pay DESC", (50,)):
    print(f"{row[0]:10} {row[1]}")

print("---")
print(conn.execute("SELECT COUNT(*), SUM(pay), AVG(pay) FROM crew").fetchone())

print("---")
for role, count, total in conn.execute(
    "SELECT role, COUNT(*), SUM(pay) FROM crew GROUP BY role ORDER BY SUM(pay) DESC"
):
    print(f"{role:10} {count} people, {total} total")

Elaine     250
Guybrush   100
---
(4, 435, 108.75)
---
governor   1 people, 250 total
captain    1 people, 100 total
lookout    2 people, 85 total

Rows as dictionaries, which is far nicer

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row          # the one line worth remembering
conn.execute("CREATE TABLE crew (name TEXT, role TEXT, pay INTEGER)")
conn.execute("INSERT INTO crew VALUES ('Guybrush', 'captain', 100)")

row = conn.execute("SELECT * FROM crew").fetchone()

print(row["name"], row["pay"])
print(dict(row))
print(row.keys())

Guybrush 100
{'name': 'Guybrush', 'role': 'captain', 'pay': 100}
['name', 'role', 'pay']

Relationships: the actual point of SQL

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")      # SQLite needs asking, every connection

conn.executescript("""
    CREATE TABLE ships (
        id   INTEGER PRIMARY KEY,
        name TEXT NOT NULL UNIQUE
    );
    CREATE TABLE crew (
        id      INTEGER PRIMARY KEY,
        name    TEXT NOT NULL,
        ship_id INTEGER REFERENCES ships(id) ON DELETE CASCADE
    );
    INSERT INTO ships (name) VALUES ('Sea Monkey'), ('Flying Dutchman');
    INSERT INTO crew (name, ship_id) VALUES
        ('Guybrush', 1), ('Otis', 1), ('LeChuck', 2);
""")

for row in conn.execute("""
    SELECT ships.name AS ship, COUNT(crew.id) AS crew_count
    FROM ships
    LEFT JOIN crew ON crew.ship_id = ships.id
    GROUP BY ships.id
    ORDER BY crew_count DESC
"""):
    print(f"{row['ship']:16} {row['crew_count']} crew")

print("---")
for row in conn.execute("""
    SELECT crew.name, ships.name AS ship
    FROM crew JOIN ships ON crew.ship_id = ships.id
    WHERE ships.name = ?
""", ("Sea Monkey",)):
    print(f"{row['name']} sails the {row['ship']}")

Sea Monkey       2 crew
Flying Dutchman  1 crew
---
Guybrush sails the Sea Monkey
Otis sails the Sea Monkey

A JOIN answers a question that would otherwise be two loops and a dictionary in Python, and the database does it with an index instead of scanning. This is what SQL is for, and it is why "just use a JSON file" stops working around a few thousand records.

Transactions: all or nothing

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE accounts (name TEXT PRIMARY KEY, balance INTEGER)")
conn.executemany("INSERT INTO accounts VALUES (?, ?)",
                 [("Guybrush", 100), ("Elaine", 250)])
conn.commit()


def transfer(conn, sender, recipient, amount):
    """Move money. Either both sides happen or neither does."""
    try:
        with conn:          # commits on success, rolls back on any exception
            balance = conn.execute(
                "SELECT balance FROM accounts WHERE name = ?", (sender,)
            ).fetchone()[0]
            if balance < amount:
                raise ValueError(f"{sender} has only {balance}")
            conn.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?",
                         (amount, sender))
            conn.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?",
                         (amount, recipient))
        return "ok"
    except ValueError as err:
        return f"refused: {err}"


print(transfer(conn, "Guybrush", "Elaine", 50))
print(transfer(conn, "Guybrush", "Elaine", 500))
print(dict(conn.execute("SELECT name, balance FROM accounts ORDER BY name").fetchall()))

ok
refused: Guybrush has only 50
{'Elaine': 300, 'Guybrush': 50}

with conn: is a transaction. If anything inside raises, every change is undone. Without it, a crash between the two UPDATE statements would destroy money, which is the classic illustration of why databases have transactions at all.

Indexes: the difference between instant and unusable

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE events (id INTEGER PRIMARY KEY, user_id INTEGER, action TEXT)")
conn.executemany("INSERT INTO events (user_id, action) VALUES (?, ?)",
                 [(i % 100, "click") for i in range(20000)])

plan_before = conn.execute(
    "EXPLAIN QUERY PLAN SELECT * FROM events WHERE user_id = 42"
).fetchone()[-1]

conn.execute("CREATE INDEX idx_events_user ON events(user_id)")

plan_after = conn.execute(
    "EXPLAIN QUERY PLAN SELECT * FROM events WHERE user_id = 42"
).fetchone()[-1]

print("before:", plan_before)
print("after: ", plan_after)

before: SCAN events
after:  SEARCH events USING INDEX idx_events_user (user_id=?)

SCAN means it read every row. SEARCH USING INDEX means it jumped straight there. On twenty thousand rows the difference is small; on twenty million it is the difference between a page that loads and a page that times out. Index the columns you filter and join on, and EXPLAIN QUERY PLAN tells you whether it worked.

Beyond SQLite

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")

conn.executescript("""
    CREATE TABLE authors (
        id   INTEGER PRIMARY KEY,
        name TEXT NOT NULL UNIQUE
    );
    CREATE TABLE books (
        id        INTEGER PRIMARY KEY,
        title     TEXT NOT NULL,
        year      INTEGER,
        author_id INTEGER NOT NULL REFERENCES authors(id)
    );
""")

conn.executemany("INSERT INTO authors (name) VALUES (?)",
                 [("Herbert",), ("Gibson",), ("Unpublished",)])
conn.executemany("INSERT INTO books (title, year, author_id) VALUES (?, ?, ?)", [
    ("Dune", 1965, 1),
    ("Dune Messiah", 1969, 1),
    ("Neuromancer", 1984, 2),
])
conn.commit()

for row in conn.execute("""
    SELECT authors.name, COUNT(books.id) AS n, MIN(books.year) AS first
    FROM authors
    LEFT JOIN books ON books.author_id = authors.id
    GROUP BY authors.id
    ORDER BY n DESC, authors.name
"""):
    first = row["first"] or "-"
    print(f"{row['name']:12} {row['n']} books, first {first}")

Herbert      2 books, first 1965
Gibson       1 books, first 1984
Unpublished  0 books, first -

def find_user(conn, username):
    return conn.execute(
        f"SELECT * FROM users WHERE username = '{username}'"
    ).fetchall()

def find_user(conn, username):
    """Find a user by name. The value is passed as data, never as SQL."""
    return conn.execute(
        "SELECT * FROM users WHERE username = ?", (username,)
    ).fetchall()