Queries You Can Trust

The analyzer needs a search: “show me every record whose message mentions this text”. It’s the kind of function that gets written in two minutes with an f-string, and it works on the first thing anyone tries.

Bug hunt: the search that broke on a word

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE records (level TEXT, source TEXT, message TEXT)")
with conn:
    conn.executemany("INSERT INTO records VALUES (?, ?, ?)", [
        ("INFO", "auth", "user=ada action=login"),
        ("ERROR", "payments", 'msg="card declined"'),
        ("ERROR", "db", "msg=\"can't reach replica\""),
        ("INFO", "auth", "user=grace action=logout"),
    ])


def search(text):
    sql = f"SELECT source, message FROM records WHERE message LIKE '%{text}%'"
    return conn.execute(sql).fetchall()


print(search("declined"))
print(search("can't"))

Reproduce. search("declined") works. search("can't") raises OperationalError: near "t": syntax error. The smallest input that fails is one word with an apostrophe.

Inspect. The error is about the SQL, so look at the SQL. Add print(sql) inside search, before the execute:

SELECT source, message FROM records WHERE message LIKE '%can't%'

Narrow. In SQL, ' starts and ends a string. The apostrophe in can't ends the string after can, and SQLite is left trying to understand t%' as more SQL. The function takes text from a user and pastes it into code. So the user isn’t typing a search term. They’re writing part of your program.

That’s bad enough when it’s an accident. Here it is on purpose:

print(len(search("' OR 1=1 --")))

That prints 4: every row in the table. The input closes the string, adds OR 1=1, which is true for every row, and -- turns the rest of the line into an SQL comment. This is SQL injection, and it’s how attackers read, change or delete data they were never meant to reach. It has been near the top of every list of common security flaws for over twenty years, and it’s entirely avoidable.

Pin it. Two tests: text with a quote in it is found, and SQL in the search text is treated as text.

def test_search_text_with_a_quote(self):
    self.assertEqual([source for source, _ in search("can't")], ["db"])

def test_search_is_not_sql(self):
    self.assertEqual(search("' OR 1=1 --"), [])

Fix. Never put a value into SQL yourself. Put a ? where the value goes, and pass the value separately:

def search(text):
    sql = "SELECT source, message FROM records WHERE message LIKE ?"
    return conn.execute(sql, (f"%{text}%",)).fetchall()

Make that change and run it again. Now sqlite3 sends the SQL and the value to SQLite separately. The SQL is fixed when you write it, and no value can ever become part of it, whatever characters it contains. Note that the % wildcards are part of the value, not the SQL.

Placeholders stop the text from being read as SQL. They don’t stop LIKE reading it as a pattern. Inside a LIKE value, % still means “anything” and _ means “any one character”, so a search for 100% also finds 1000. For a literal search, escape those characters and tell SQLite which character does the escaping:

def like_pattern(text):
    """A LIKE pattern matching text anywhere, with LIKE's own wildcards made literal."""
    for special in ("\\", "%", "_"):                   # the escape character goes first
        text = text.replace(special, "\\" + special)
    return f"%{text}%"


conn.execute("SELECT ... WHERE message LIKE ? ESCAPE '\\'", (like_pattern(text),))

The analyzer’s search does this from the Day 3 reference on. LIKE also ignores case for ASCII letters, which is what a search box usually wants.

Placeholders in practice

The value parameters are always a sequence, even for one value. (text,) with its trailing comma is a one-item tuple. (text) is just text in brackets, and sqlite3 would treat each character of the string as a separate value, and complain that there are more values than placeholders.

With several values, named placeholders are easier to read than a row of ?s, and the parameters become a dict:

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE records (timestamp TEXT, level TEXT, source TEXT)")
with conn:
    conn.executemany("INSERT INTO records VALUES (?, ?, ?)", [
        ("2026-09-01 09:15:30", "ERROR", "payments"),
        ("2026-09-01 09:40:00", "ERROR", "search"),
        ("2026-09-01 10:02:11", "ERROR", "payments"),
        ("2026-09-01 10:15:00", "INFO", "payments"),
    ])

query = """
    SELECT timestamp, source FROM records
    WHERE level = :level AND timestamp >= :start AND timestamp < :end
    ORDER BY timestamp
"""
params = {"level": "ERROR", "start": "2026-09-01 09:30:00", "end": "2026-09-01 11:00:00"}
for row in conn.execute(query, params):
    print(row)

What placeholders can’t do

A placeholder stands for a value. It can’t stand for a table name, a column name, a keyword like DESC, or a piece of SQL. So what about a function that counts records per level or per source?

count_by(conn, "level")    # {'ERROR': 5, 'INFO': 2, ...}
count_by(conn, "source")   # {'payments': 4, 'search': 1, ...}

GROUP BY ? doesn’t do what you’d want. It groups by the constant string 'level', so every row lands in the same group. The column name has to be part of the SQL text. That’s safe only if the name can’t come from outside the program, and the way to guarantee it is to check it against a fixed set of allowed names first:

COUNTABLE = {"level", "source"}


def count_by(conn, column):
    if column not in COUNTABLE:     # a column name can't be a ? parameter
        raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}")
    sql = f"SELECT {column}, COUNT(*) FROM records GROUP BY {column}"
    return dict(conn.execute(sql).fetchall())

The f-string is safe here because column can only be one of two strings you wrote yourself. This is an allow-list: you list what’s permitted, and refuse everything else. The opposite, trying to spot and reject dangerous characters, is a losing game. There’s always a character you didn’t think of.

Your turn

import sqlite3
import sys
import unittest


def count_by(conn, column, level=None):
    """Count records per level or per source, busiest first, optionally for one level."""
    raise NotImplementedError("your code here")


class TestCountBy(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        self.conn.execute("CREATE TABLE records (level TEXT, source TEXT)")
        self.conn.executemany("INSERT INTO records VALUES (?, ?)", [
            ("ERROR", "payments"), ("ERROR", "search"), ("INFO", "auth"),
            ("ERROR", "payments"), ("WARNING", "payments"), ("INFO", "search"),
        ])

    def test_by_level(self):
        self.assertEqual(count_by(self.conn, "level"), {"ERROR": 3, "INFO": 2, "WARNING": 1})

    def test_busiest_first_then_alphabetical(self):
        self.assertEqual(list(count_by(self.conn, "source")), ["payments", "search", "auth"])

    def test_one_level_only(self):
        self.assertEqual(count_by(self.conn, "source", level="ERROR"),
                         {"payments": 2, "search": 1})

    def test_unknown_column_is_refused(self):
        with self.assertRaises(ValueError):
            count_by(self.conn, "source FROM records; --")

    def test_level_is_a_value_not_sql(self):
        self.assertEqual(count_by(self.conn, "level", level="x' OR '1'='1"), {})


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestCountBy)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

COUNTABLE = {"level", "source"}


def count_by(conn, column, level=None):
    """Count records per level or per source, busiest first, optionally for one level."""
    if column not in COUNTABLE:
        raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}")
    where, params = ("WHERE level = ?", (level,)) if level else ("", ())
    sql = (f"SELECT {column}, COUNT(*) AS n FROM records {where} "
           f"GROUP BY {column} ORDER BY n DESC, {column}")
    return dict(conn.execute(sql, params).fetchall())

What you can now do

• Recognise SQL injection, both the accidental kind and the deliberate kind.
• Pass every value with ? or named :placeholders, including the % of a LIKE pattern.
• Put a column name into SQL only after checking it against an allow-list.

Next: moving the analyzer’s storage into its own module, and testing it against a database that exists only for the length of one test.