Pebellslearn by building
0%
RoadmapsMatrixCoursesSandboxSign in
Serialisation & Storage  ›  Lesson

sqlite3 & the DB-API

SQLite 26 minConnections, cursors, parameters and transactions: getting records into a database and back unchanged
You're building a piece ofQuarry — the vault
This piece — store_batch(): Puts a batch into SQLite in one transaction: all of it, or none of it.
Scenario A batch of parsed incidents is ready. One source file was named by someone creative, and one record repeats an id. The vault must keep the first safe and refuse the second batch whole.
Your task
Build store_batch(raws): open a :memory: database with row_factory = sqlite3.Row, create SCHEMA, executemany the batch with :id/:source/:level/:ms placeholders inside `with conn:`, record "IntegrityError" if the batch is refused, then return {"rows": every row as a dict in id order, "error": None or "IntegrityError"}. Example: a batch with a duplicate id → {"rows": [], "error": "IntegrityError"}.

sqlite3 & the DB-API

In short: CSV, JSON and pickle all write the whole thing and read the whole thing — fine for a batch, hopeless for a question like which sources logged an ERROR yesterday. SQLite is a complete SQL database in a single file (or entirely in memory), built into Python, and sqlite3 follows the DB-API: connect, get a cursor, execute, fetch — the same shape every other Python database driver uses. The one rule that matters most: values travel into SQL as parameters (? or :name), never pasted into the query text, because a pasted value can rewrite what the query means. A transaction makes a batch of inserts succeed or fail together, so a bad record 312 of 500 never leaves 311 silently stored.

You'll build store_batch(raws), the vault itself: a batch goes in as one transaction, all of it or none of it.

The value nobody' OR '1'='1 pasted into SQL text rewrites the WHERE clause so every row matches; the same value bound as a ? parameter travels as data and matches nothing

Look up a customer by a name with an apostrophe

The obvious way to put a value into a query is an f-string, and it's the one way you must never use — it breaks on an ordinary surname before anyone attacks anything:

What happens when this runs?
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE people (name TEXT)")
conn.execute("INSERT INTO people VALUES ('Siobhan')")

name = "O'Brien"
rows = conn.execute(f"SELECT * FROM people WHERE name = '{name}'").fetchall()
print(len(rows))

Fix the lookup so a real surname works, using a parameter instead of a pasted string:

broken — fix itOperationalError

Run it, then rewrite the SELECT so the name travels as a parameter.

Stop a hostile "source name" from rewriting the query

That's the accidental version. The deliberate version is SQL injection: a value crafted to change what the query means without breaking its syntax:

python

The formatted query returns every row, because the value rewrote the WHERE clause into something always true. The parameterised query returns none, because it looked for a source literally named nobody' OR '1'='1, and there isn't one. That difference is structural: with ?, SQL text and values travel separately — the statement compiles first and the value is bound afterwards as data, so it can never become SQL, whatever characters it holds. Two placeholder styles work: ? with a tuple, and :name with a dict — handy because Quarry's records already are dicts:

python
Drawing diagram…

Placeholders stand for values only — you can't parameterise a table or column name. If one really has to vary, choose it from a fixed allow-list in your code, never straight from input.

Insert a batch where one row breaks a rule

A batch of 500 records where record 312 duplicates a primary key should not leave 311 stored — the next run couldn't tell what landed and what didn't. with conn: commits if the block finishes and rolls back if it raises:

What does this print?
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE records (id INTEGER PRIMARY KEY, ms INTEGER)")
batch = [(1, 120), (2, 95), (1, 40)]          # id 1 appears twice

try:
    with conn:
        conn.executemany("INSERT INTO records VALUES (?, ?)", batch)
except sqlite3.IntegrityError:
    pass

print(conn.execute("SELECT COUNT(*) FROM records").fetchone()[0])

Constraints are what make "bad" detectable: PRIMARY KEY refuses duplicates, NOT NULL refuses missing values, CHECK (level IN (...)) refuses a level Quarry doesn't know — each raises sqlite3.IntegrityError at insert time, instead of turning up in a report next week. executemany takes one statement and an iterable of parameter sets, and conn.row_factory = sqlite3.Row lets a row behave like both a tuple and a mapping:

step through it
1import sqlite3
2 
3conn = sqlite3.connect(":memory:")
4conn.row_factory = sqlite3.Row
5conn.execute("CREATE TABLE records (id INTEGER PRIMARY KEY, level TEXT, ms INTEGER)")
6batch = [{"id": 1, "level": "WARN", "ms": 120}, {"id": 2, "level": "INFO", "ms": 95}]
7conn.executemany("INSERT INTO records VALUES (:id, :level, :ms)", batch)
8rows = []
9for row in conn.execute("SELECT * FROM records ORDER BY id"):
10 rows.append(dict(row))
11print(rows)

Traps worth knowing

SQLite stores five kinds of value: NULL, integer, real, text, blob — None, int, float, str, bytes. Anything else needs a decision, the same kind JSON forced on you: a datetime becomes value.isoformat() as TEXT (sqlite3's automatic datetime conversion is deprecated since 3.12 — be explicit); a list or set, like Quarry's tags, becomes json.dumps(sorted(tags)) as TEXT; a Decimal becomes str(value).

SQLite doesn't enforce column types by default. A column declared INTEGER will quietly store the string "slow" unless you add STRICT after the table definition, a CHECK constraint, or convert before inserting.

Quick practice

What does this print?
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, note TEXT NOT NULL)")
try:
    conn.execute("INSERT INTO t (id) VALUES (1)")
except sqlite3.IntegrityError as exc:
    print(type(exc).__name__)
broken — fix itProgrammingError

Run it and read which type the driver refuses to bind. Pack the list as JSON text before inserting.

What does this print?
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (n INTEGER)")
with conn:
    conn.execute("INSERT INTO t VALUES (?)", (5,))

try:
    with conn:
        conn.execute("UPDATE t SET n = n + 1")
        raise ValueError("boom")
except ValueError:
    pass
print(conn.execute("SELECT n FROM t").fetchone()[0])

🎯 Your turn

Write store_batch(raws). Each raw record has id, source, level and ms.

  1. Connect to ":memory:", set row_factory = sqlite3.Row, and create the table with the starter's SCHEMA.
  2. Insert the whole batch with executemany and :name placeholders, inside with conn:.
  3. If that raises sqlite3.IntegrityError, note the error's class name. The with block has already rolled back.
  4. Read everything back with SELECT id, source, level, ms FROM records ORDER BY id, turning each row into a dict.
  5. Return {"rows": [...], "error": None}, or with "error": "IntegrityError" when the batch was refused, in which case rows is [].

A source named web'); DROP TABLE records; -- must be stored exactly as written.

Hint: set error = None before the try, and do the SELECT after the try/except, so both paths share it.

Press ▶ Run and hit ✓ Check. ✅

Practice — level up

0/6 solved

Solve each in its own editor — Run to try, Check to grade. Stuck? Reveal a Hint, or the full Solution + walkthrough.

Warm-upIn, then count

Write count_rows(values). Make a :memory: database with a table t (v), insert each value as its own row with executemany and a ? placeholder, and return SELECT COUNT(*) FROM t as an int.

sqlitesqlite3
DrillA parameterised lookup

Write find_by_level(raws, level). Store the records (each has id, level, ms) in a :memory: table, then return the ids whose level equals level, in id order, using a ? parameter. The last test passes an injection attempt as the level: it must match nothing.

sqliteparameterized-query
BuildFormatted versus parameterised, side by side

The starter fills a people table with Ada, O'Brien and Siobhan. Write injection_demo(name) that runs the same lookup twice: once with an f-string (WHERE name = '{name}') and once with a ? parameter. Return {"formatted": row count or the exception's class name, "param": row count}.

sqlitesql-injection
BossOne transaction per batch

Write load_batches(batches). Each batch is a list of [id, ms] pairs. Use one :memory: table records (id INTEGER PRIMARY KEY, ms INTEGER NOT NULL) for all of them, and insert each batch in its own with conn: block. A batch that raises sqlite3.IntegrityError is rolled back on its own. Return {"stored": total rows in the table, "failed": [indexes of failed batches]}.

sqlitetransaction
CapstoneReruns that don't duplicate

Quarry reruns a batch whenever a source file changes, so storing must be idempotent. Write upsert_runs(first, second): create records (id INTEGER PRIMARY KEY, level TEXT NOT NULL, ms INTEGER), then store first and then second, each in its own transaction, with INSERT ... ON CONFLICT(id) DO UPDATE SET level = excluded.level, ms = excluded.ms. Return all rows as dicts (id, level, ms) in id order.

sqlitetransactionparameterized-query
ExplainWhy `?` beats escaping quotes yourself

A teammate proposes fixing the f-string queries by writing name.replace("'", "''") before formatting them in. Explain what the ? placeholder does differently, why that makes it safe in a way escaping isn't, and why the O'Brien crash matters even though nobody was attacking.

store_batch.py
Call the function you wrote — just like your app's frontend would. Edit the inputs and hit Call.
store_batch(raws) → dictStore a batch in one transaction and read the vault back.
try:
returned
Press Run to execute your code.
Press Check to run the tests for this lesson.