Records into SQLite

Warm-up: from Day 2

From memory first: the effort of recalling is what makes it stick.

1. Predict both lines.

line = '2026-09-01 09:15:30 ERROR   payments: order=1043 msg="card declined"'
date, time, level, rest = line.split(maxsplit=3)
source, _, message = rest.partition(":")

print(level, source)
print(message.strip().split()[0])

ERROR payments
order=1043

2. Fix. This should report the count and then list the errors. The list comes out empty.

def errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

found = errors(["ERROR a", "INFO b", "ERROR c"])
print(len(list(found)), "errors")
print(list(found))

def errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

found = list(errors(["ERROR a", "INFO b", "ERROR c"]))
print(len(found), "errors")
print(found)

2 errors
['ERROR a', 'ERROR c']

3. Explain. No code: why does Day 2’s parser collect bad lines in an errors list instead of raising on the first one?

Where Day 2 ended

Here’s where Day 2 ended: a streaming parser that turns lines into records and reports every line it rejects.

Every question the analyzer answers today means reading and parsing the whole log again, and when the program ends the records are gone. That’s fine for one report. It’s not fine when someone asks “how many payment errors were there last Tuesday between 9 and 10?” and the log is a gigabyte. Parse once, keep the records, and ask as many questions as you like. That’s what a database is for.

SQLite: a database in a file

SQLite is a complete SQL database that lives in a single file, with no server to install or run. Python ships with it as the sqlite3 module. It’s in every phone, every browser and most desktop apps, and it’s what’s running in this page.

import sqlite3

conn = sqlite3.connect(":memory:")     # a database in memory; use a file name to keep it

conn.execute("""
    CREATE TABLE records (
        timestamp TEXT NOT NULL,
        level     TEXT NOT NULL,
        source    TEXT NOT NULL,
        message   TEXT NOT NULL
    )
""")

rows = [
    ("2026-09-01 09:14:02", "INFO", "auth", "user=ada action=login"),
    ("2026-09-01 09:15:30", "ERROR", "payments", "order=1043"),
    ("2026-09-01 09:17:45", "ERROR", "search", "query=dune"),
    ("2026-09-01 10:02:11", "ERROR", "payments", "order=1050"),
]
with conn:                             # a transaction: committed if the block succeeds
    conn.executemany("INSERT INTO records VALUES (?, ?, ?, ?)", rows)

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

query = """
    SELECT source, COUNT(*) FROM records
    WHERE level = 'ERROR'
    GROUP BY source
    ORDER BY COUNT(*) DESC
"""
for source, count in conn.execute(query):
    print(f"{source:10} {count}")

The whole API you need today is small:

The ? in the INSERT is a placeholder. sqlite3 fills it in with the matching value, safely. The next lesson is about why you should never build SQL any other way.

Rows with names

By default each row is a tuple, so you have to remember that column 2 is the source. Set row_factory to sqlite3.Row and rows can also be read by column name:

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE records (level TEXT, source TEXT)")
conn.execute("INSERT INTO records VALUES (?, ?)", ("ERROR", "payments"))

row = conn.execute("SELECT * FROM records").fetchone()
print(row["source"], row["level"])     # by name
print(row[0], tuple(row))              # still works like a tuple
print(row.keys())

SQL in five clauses

You’ll use five clauses all week. A query reads in this order, even though the database works out the answer differently:

SELECT   source, COUNT(*) AS n       -- which columns, or calculations over them
FROM     records                     -- which table
WHERE    level = 'ERROR'             -- keep only these rows
GROUP BY source                      -- one output row per distinct source
ORDER BY n DESC                      -- sort the output

COUNT(*), MIN(...), MAX(...) and SUM(...) are aggregates. With GROUP BY they’re calculated once per group, and without it, once for the whole table. That one query replaces the Counter loop from Day 2, and it runs inside the database without building a single Python object per row.

Bug hunt: the record that came back different

SQLite has only five kinds of value: NULL, INTEGER, REAL, TEXT and BLOB. There’s no date type. Here’s a first attempt at saving LogRecords and loading them back, with a test checking the round trip:

import sqlite3
import sys
import unittest
from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class LogRecord:
    timestamp: datetime
    level: str
    source: str


def save(conn, records):
    with conn:
        conn.executemany("INSERT INTO records VALUES (?, ?, ?)",
                         [(r.timestamp, r.level, r.source) for r in records])


def load(conn):
    return [LogRecord(*row) for row in conn.execute("SELECT * FROM records")]


class TestRoundTrip(unittest.TestCase):
    def test_records_come_back_equal(self):
        conn = sqlite3.connect(":memory:")
        conn.execute("CREATE TABLE records (timestamp TEXT, level TEXT, source TEXT)")
        original = [LogRecord(datetime(2026, 9, 1, 9, 14, 2), "INFO", "auth")]
        save(conn, original)
        self.assertEqual(load(conn), original)


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

Reproduce. One record, one test, and it fails.

Inspect. Read the two records in the failure message side by side. One has timestamp='2026-09-01 09:14:02', a string. The other has timestamp=datetime.datetime(2026, 9, 1, 9, 14, 2). Same moment, but one of them is text, and text never equals a datetime. Look at the rest of the output too. There’s a DeprecationWarning: “The default datetime adapter is deprecated”.

Before you open this: why did the timestamp come back as a string? The warning is a clue. Then try to make the round trip pass. Open the reveal afterwards even if you did: it explains how sqlite3 converted the value behind your back, which matters for everything else you store.

def save(conn, records):
    with conn:
        conn.executemany("INSERT INTO records VALUES (?, ?, ?)",
                         [(r.timestamp.isoformat(sep=" "), r.level, r.source) for r in records])


def load(conn):
    return [LogRecord(datetime.fromisoformat(ts), level, source)
            for ts, level, source in conn.execute("SELECT * FROM records")]

Your turn

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE records (timestamp TEXT, level TEXT, source TEXT, message TEXT)")
with conn:
    conn.executemany("INSERT INTO records VALUES (?, ?, ?, ?)", [
        ("2026-09-01 09:14:02", "INFO", "auth", "user=ada action=login"),
        ("2026-09-01 09:14:07", "WARNING", "payments", "order=1042 retry=1"),
        ("2026-09-01 09:15:30", "ERROR", "payments", "order=1043"),
        ("2026-09-01 09:17:45", "ERROR", "search", "query=dune"),
        ("2026-09-01 10:02:11", "ERROR", "payments", "order=1050"),
        ("2026-09-01 10:05:00", "INFO", "auth", "user=alan action=login"),
        ("2026-09-01 10:31:40", "ERROR", "db", "replica lag"),
        ("2026-09-01 11:00:00", "ERROR", "payments", "order=1061"),
    ])

# 1. records per level

# 2. errors after 10:00, per source

# 3. first and last timestamp

for row in conn.execute(
        "SELECT level, COUNT(*) AS n FROM records GROUP BY level ORDER BY n DESC"):
    print(row["level"], row["n"])

for row in conn.execute("""
        SELECT source, COUNT(*) AS n FROM records
        WHERE level = 'ERROR' AND timestamp >= '2026-09-01 10:00:00'
        GROUP BY source ORDER BY n DESC"""):
    print(row["source"], row["n"])

first, last = conn.execute("SELECT MIN(timestamp), MAX(timestamp) FROM records").fetchone()
print(first, "to", last)

What you can now do

• Create a SQLite table, insert rows with executemany, and query them.
• Use with conn: to commit a group of changes together.
• Read rows by column name with sqlite3.Row.
• Write SELECT … WHERE … GROUP BY … ORDER BY with aggregates.
• Convert values like datetime explicitly on the way in and out of storage.

Next: the ? placeholders, and what goes wrong without them.