Everything the analyzer does with SQLite now goes into one module, store.py,
next to analyzer.py. Parsing stays in analyzer.py and knows nothing about
databases. Storage goes in store.py and knows nothing about log lines. It’s
the same split as Day 1’s rules and program, for the same reason: each can
change without the other.
The connection is a parameter
Every function in store.py takes the connection as its first argument:
None of them opens a database itself. That one decision is what makes the module
easy to test. The real program passes a connection to a file on disk. A test
passes a connection to a brand-new database in memory. The functions can’t tell
the difference, and don’t need to.
A single connect function creates the table, so every caller gets the same
schema:
def connect(path=":memory:"): """Open (or create) a database with the records table in it.""" conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute(SCHEMA) # CREATE TABLE IF NOT EXISTS records (...) return conn
A fresh database for every test
:memory: databases are created in microseconds and vanish when closed. So
don’t share one between tests. Make a new one in setUp:
class TestStore(unittest.TestCase): def setUp(self): self.conn = connect() # a new, empty database for every test self.addCleanup(self.conn.close)
Every test starts from an empty table, so no test can see rows another test
left behind, and they pass in any order. Day 1’s lesson on order-dependent tests
applies to databases more than anywhere else.
addCleanup registers something to run after the test, whether it passes,
fails or errors. It’s tearDown without the separate method, and you register
it right next to the thing it cleans up, so it can’t be forgotten.
What an in-memory database doesn't test
Tests against :memory: SQLite prove your SQL and your Python are right. They
don’t prove anything about a different database, like PostgreSQL in
production, which has its own dialect and its own behaviour. The usual answer is
a few extra tests against the real thing, run less often. For an SQLite-backed
tool like this one, in-memory tests are the real thing.
Bug hunt: every error, twice
The analyzer imports a log into the database every night. Last night the import
timed out, and the scheduler did what schedulers do: it ran the job again.
An import that ran twicePython"""The nightly import. Last night it timed out, and the scheduler retried it."""
from analyzer import parse_lines, read_lines
from store import connect, count_by, save_records
conn = connect()
for attempt in (1, 2):
stored = save_records(conn, "messy.log", parse_lines(read_lines("messy.log"), []))
print(f"attempt {attempt}: stored {stored} records")
print("Levels:", count_by(conn, "level"))
"""Log analyzer, Day 3: keep parsed records in SQLite, and ask questions in SQL."""
import sqlite3
from datetime import datetime
from analyzer import LogRecord, parse_fields
SCHEMA = """
CREATE TABLE IF NOT EXISTS records (
log TEXT NOT NULL,
line_number INTEGER NOT NULL,
timestamp TEXT NOT NULL, -- ISO 8601, so text order is time order
level TEXT NOT NULL,
source TEXT NOT NULL,
message TEXT NOT NULL
)
"""
def connect(path=":memory:"):
"""Open (or create) a database with the records table in it."""
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute(SCHEMA)
return conn
def save_records(conn, log, records):
"""Store records read from one log. Returns how many were stored."""
rows = ((log, r.line_number, r.timestamp.isoformat(sep=" "), r.level, r.source, r.message)
for r in records)
with conn: # one transaction: all rows, or none
cursor = conn.executemany(
"INSERT INTO records VALUES (?, ?, ?, ?, ?, ?)", rows)
return cursor.rowcount
def record_from_row(row):
"""Rebuild a LogRecord. Fields are re-parsed from the message, not stored twice."""
return LogRecord(
timestamp=datetime.fromisoformat(row["timestamp"]),
level=row["level"],
source=row["source"],
message=row["message"],
fields=parse_fields(row["message"]),
line_number=row["line_number"],
)
def load_records(conn, log):
rows = conn.execute(
"SELECT * FROM records WHERE log = ? ORDER BY line_number", (log,))
return [record_from_row(row) for row in rows]
def search(conn, text):
"""Records whose message contains text, oldest first."""
rows = conn.execute(
"SELECT * FROM records WHERE message LIKE ? ORDER BY timestamp", (f"%{text}%",))
return [record_from_row(row) for row in rows]
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: # a column name can't be a ? parameter
raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}")
where, params = ("WHERE level = ?", (level,)) if level else ("", ())
rows = conn.execute(
f"SELECT {column} AS key, COUNT(*) AS n FROM records {where} "
f"GROUP BY {column} ORDER BY n DESC, key", params)
return {row["key"]: row["n"] for row in rows}
"""Log analyzer, end of Day 2: lines become records, streamed and validated."""
import re
from collections import Counter
from dataclasses import dataclass, field
from datetime import datetime
LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
# key=value, where a value is either "quoted, spaces allowed" or a single word.
FIELD = re.compile(r'(?P<key>\w+)=(?P<value>"[^"]*"|\S+)')
@dataclass(frozen=True)
class LogRecord:
timestamp: datetime
level: str
source: str
message: str
fields: dict[str, str] = field(default_factory=dict)
line_number: int = 0
class ParseError(ValueError):
"""A line that is not a valid log record."""
def __init__(self, line_number, reason, line):
super().__init__(f"line {line_number}: {reason}: {line!r}")
self.line_number = line_number
self.reason = reason
self.line = line
def parse_fields(text):
"""Turn 'a=1 b="two words"' into {'a': '1', 'b': 'two words'}."""
return {m["key"]: m["value"].strip('"') for m in FIELD.finditer(text)}
def parse_record(line, line_number=0):
"""Parse one line, or raise ParseError saying what is wrong with it."""
parts = line.split(maxsplit=3)
if len(parts) < 4:
raise ParseError(line_number, "too few parts", line)
date, time, level, rest = parts
try:
timestamp = datetime.fromisoformat(f"{date} {time}")
except ValueError:
raise ParseError(line_number, "bad timestamp", line) from None
if level not in LEVELS:
raise ParseError(line_number, f"unknown level {level!r}", line)
source, colon, message = rest.partition(":")
if not colon:
raise ParseError(line_number, "missing 'source:'", line)
message = message.strip()
return LogRecord(timestamp, level, source, message, parse_fields(message), line_number)
def read_lines(path):
"""Yield the lines of a file one at a time, without line endings."""
with open(path) as file:
for line in file:
yield line.rstrip("\n")
def parse_lines(lines, errors):
"""Yield a record per good line; add a ParseError to the errors list per bad one."""
for number, line in enumerate(lines, start=1):
if not line.strip():
continue
try:
yield parse_record(line, number)
except ParseError as error:
errors.append(error)
def summarize(records):
"""Count levels and errors per source in a single pass."""
levels, error_sources = Counter(), Counter()
for record in records:
levels[record.level] += 1
if record.level == "ERROR":
error_sources[record.source] += 1
return levels, error_sources
2026-09-01 09:14:02 INFO auth: user=ada action=login
2026-09-01 09:14:07 WARNING payments: order=1042 amount=19.99 retry=1
2026-09-01 09:15:30 ERROR payments: order=1043 msg="card declined"
2026-09-01 09:16:11 INFO auth: user=alan action=logout
--- rotated at 09:16 ---
2026-09-01 09:17:45 ERROR search: query="python books" msg="index offline"
2026-09-01 9:18 INFO search: query=dune results=4
2026-09-01 09:19:30 ERROR auth: user=grace msg="token expired"
Reproduce. Two attempts, six records each, and the level counts are exactly
double what messy.log contains. Six ERRORs from a log that has three.
Inspect. The state that matters is the table itself. Ask it whether any line
was stored more than once. Add this to the end of main.py:
query = """ SELECT log, line_number, COUNT(*) AS copies FROM records GROUP BY log, line_number HAVING copies > 1"""for row in conn.execute(query): print(tuple(row))
Every line is there twice. (HAVING is WHERE for groups: it filters afterGROUP BY has counted.)
Before you open this, try to change store.py so the second attempt stores nothing. Read the reveal either way: it introduces primary keys and ON CONFLICT, which the rest of this lesson and the next three days rely on.
The diagnosis and the fix (read it either way)
Narrow. Open store.py. The table has no way to tell that two rows are the
same line. There’s nothing marking (log, line_number) as unique, so storing
the same log again just adds more rows. The code is correct for one run and
wrong for two, and “run it again” is the first thing anyone does when a job
fails.
Pin it. A test that saves the same records twice and checks nothing changed
the second time:
Fix. Two changes to store.py. First, tell the database what makes a row
unique, by adding a primary key to the table:
message TEXT NOT NULL, PRIMARY KEY (log, line_number)
Then tell the INSERT what to do when a row with that key already exists:
cursor = conn.executemany( "INSERT INTO records VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT (log, line_number) DO NOTHING", rows)return cursor.rowcount
cursor.rowcount after executemany is the number of rows actually inserted,
so save_records can report how many were new. Run it: the second attempt
stores 0.
An operation you can safely repeat is called idempotent. Anything that runs
on a schedule, or gets retried after a timeout, should be.
The key (log, line_number) makes one assumption worth stating: a log’s name
keeps meaning the same content. If server.log is rotated and a new file takes
the old name, its line 1 is a different line, and it would be skipped as a
duplicate. A long-lived store would identify each file by something that doesn’t
get reused, like its name plus the date it was rotated. For this course, one
name means one file.
Why not INSERT OR IGNORE?
You’ll often see INSERT OR IGNORE used for this. It ignores more than
duplicates: UNIQUE, PRIMARY KEY, CHECK and NOT NULL failures are all
skipped. (Foreign-key failures are not.) A row with a missing level, which
NOT NULL should have rejected loudly, gets silently skipped instead. That’s
Day 2’s silent drop, happening inside the database. ON CONFLICT (…) DO NOTHING
names the one conflict you expect, and lets every other error through.
All or nothing
save_records wraps its insert in with conn:. That makes the whole save one
transaction. If any row fails, every row from that call is rolled back, so
the table never holds half an import. You’d otherwise have to work out which
lines made it in before you could safely retry.
That’s a promise the module makes, so it deserves a test, and writing that test
is your exercise.
Your turn
Exercise
Write two tests for store.py
test_store.py has one finished test and two that just call self.fail.
Replace each self.fail(...) with a real test:
test_search_results_are_oldest_first: LINES is deliberately out of
time order. Check that search returns matching records sorted by time,
oldest first.
test_a_failed_save_stores_nothing: call save_records with three
records where the middle one is invalid, check that it raises
sqlite3.IntegrityError, and check that the table is still empty afterwards.
For an invalid record, dataclasses.replace(record, level=None) makes a copy
with a None level, which the table’s NOT NULL will refuse.
Then check that your second test really tests something. In store.py, remove
the with conn: line (and dedent the line under it). Your test should now fail.
Put it back afterwards.
import unittest
from analyzer import parse_record
from store import connect, count_by, save_records, search
LINES = [
"2026-09-01 10:02:11 ERROR db: msg=\"can't reach replica\"",
"2026-09-01 09:14:02 INFO auth: user=ada action=login",
"2026-09-01 09:15:30 ERROR payments: order=1043 msg=\"card declined\"",
]
def records():
return [parse_record(line, number) for number, line in enumerate(LINES, start=1)]
class TestStore(unittest.TestCase):
def setUp(self):
self.conn = connect() # a new, empty database for every test
self.addCleanup(self.conn.close)
def test_saving_the_same_log_twice_stores_it_once(self):
self.assertEqual(save_records(self.conn, "a.log", records()), 3)
self.assertEqual(save_records(self.conn, "a.log", records()), 0)
def test_search_results_are_oldest_first(self):
save_records(self.conn, "a.log", records())
self.fail("write this test")
def test_a_failed_save_stores_nothing(self):
self.fail("write this test")
"""Log analyzer, Day 3: keep parsed records in SQLite, and ask questions in SQL."""
import sqlite3
from datetime import datetime
from analyzer import LogRecord, parse_fields
SCHEMA = """
CREATE TABLE IF NOT EXISTS records (
log TEXT NOT NULL,
line_number INTEGER NOT NULL,
timestamp TEXT NOT NULL, -- ISO 8601, so text order is time order
level TEXT NOT NULL,
source TEXT NOT NULL,
message TEXT NOT NULL,
PRIMARY KEY (log, line_number)
)
"""
def connect(path=":memory:"):
"""Open (or create) a database with the records table in it."""
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute(SCHEMA)
return conn
def save_records(conn, log, records):
"""Store records read from one log. Lines already stored are skipped.
Returns how many new records were stored.
"""
rows = ((log, r.line_number, r.timestamp.isoformat(sep=" "), r.level, r.source, r.message)
for r in records)
with conn: # one transaction: all rows, or none
cursor = conn.executemany(
"INSERT INTO records VALUES (?, ?, ?, ?, ?, ?) "
"ON CONFLICT (log, line_number) DO NOTHING", rows)
return cursor.rowcount
def record_from_row(row):
"""Rebuild a LogRecord. Fields are re-parsed from the message, not stored twice."""
return LogRecord(
timestamp=datetime.fromisoformat(row["timestamp"]),
level=row["level"],
source=row["source"],
message=row["message"],
fields=parse_fields(row["message"]),
line_number=row["line_number"],
)
def load_records(conn, log):
rows = conn.execute(
"SELECT * FROM records WHERE log = ? ORDER BY line_number", (log,))
return [record_from_row(row) for row in rows]
def search(conn, text):
"""Records whose message contains text, oldest first."""
rows = conn.execute(
"SELECT * FROM records WHERE message LIKE ? ORDER BY timestamp", (f"%{text}%",))
return [record_from_row(row) for row in rows]
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: # a column name can't be a ? parameter
raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}")
where, params = ("WHERE level = ?", (level,)) if level else ("", ())
rows = conn.execute(
f"SELECT {column} AS key, COUNT(*) AS n FROM records {where} "
f"GROUP BY {column} ORDER BY n DESC, key", params)
return {row["key"]: row["n"] for row in rows}
"""Log analyzer, end of Day 2: lines become records, streamed and validated."""
import re
from collections import Counter
from dataclasses import dataclass, field
from datetime import datetime
LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
# key=value, where a value is either "quoted, spaces allowed" or a single word.
FIELD = re.compile(r'(?P<key>\w+)=(?P<value>"[^"]*"|\S+)')
@dataclass(frozen=True)
class LogRecord:
timestamp: datetime
level: str
source: str
message: str
fields: dict[str, str] = field(default_factory=dict)
line_number: int = 0
class ParseError(ValueError):
"""A line that is not a valid log record."""
def __init__(self, line_number, reason, line):
super().__init__(f"line {line_number}: {reason}: {line!r}")
self.line_number = line_number
self.reason = reason
self.line = line
def parse_fields(text):
"""Turn 'a=1 b="two words"' into {'a': '1', 'b': 'two words'}."""
return {m["key"]: m["value"].strip('"') for m in FIELD.finditer(text)}
def parse_record(line, line_number=0):
"""Parse one line, or raise ParseError saying what is wrong with it."""
parts = line.split(maxsplit=3)
if len(parts) < 4:
raise ParseError(line_number, "too few parts", line)
date, time, level, rest = parts
try:
timestamp = datetime.fromisoformat(f"{date} {time}")
except ValueError:
raise ParseError(line_number, "bad timestamp", line) from None
if level not in LEVELS:
raise ParseError(line_number, f"unknown level {level!r}", line)
source, colon, message = rest.partition(":")
if not colon:
raise ParseError(line_number, "missing 'source:'", line)
message = message.strip()
return LogRecord(timestamp, level, source, message, parse_fields(message), line_number)
def read_lines(path):
"""Yield the lines of a file one at a time, without line endings."""
with open(path) as file:
for line in file:
yield line.rstrip("\n")
def parse_lines(lines, errors):
"""Yield a record per good line; add a ParseError to the errors list per bad one."""
for number, line in enumerate(lines, start=1):
if not line.strip():
continue
try:
yield parse_record(line, number)
except ParseError as error:
errors.append(error)
def summarize(records):
"""Count levels and errors per source in a single pass."""
levels, error_sources = Counter(), Counter()
for record in records:
levels[record.level] += 1
if record.level == "ERROR":
error_sources[record.source] += 1
return levels, error_sources
import sys
import unittest
import test_store
suite = unittest.defaultTestLoader.loadTestsFromModule(test_store)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Hint 1: the idea
A good test is one that would fail if the code were wrong. For the order test,
that means a search term matching records stored in the wrong order — if they
were already in order, the test would pass even with no sorting at all. For the
transaction test, the bad record has to come after a good one, so there’s
something that needs rolling back.
Hint 2: a first step
For test 1, read LINES and find a word that appears in two lines whose times
are out of order; save, search, and compare the list of sources with the order
you expect. For test 2, put replace(good[1], level=None) in the middle of
three records, and wrap the save in with self.assertRaises(...).
Show one solution
import sqlite3from dataclasses import replace... def test_search_results_are_oldest_first(self): save_records(self.conn, "a.log", records()) found = search(self.conn, "msg=") self.assertEqual([r.source for r in found], ["payments", "db"]) def test_a_failed_save_stores_nothing(self): good = records() bad = replace(good[1], level=None) # NOT NULL: the database refuses it with self.assertRaises(sqlite3.IntegrityError): save_records(self.conn, "a.log", [good[0], bad, good[2]]) self.assertEqual(count_by(self.conn, "level"), {})
The search test picks a term, msg=, that matches two records stored in the
wrong order (the db line comes first in LINES but happened later). A test
where the input is already sorted would pass even if search didn’t sort at
all.
The transaction test puts the bad record in the middle. The first record is
inserted, then the second fails. Without with conn:, that first row stays in
the connection’s open transaction and shows up in the count. That’s the failure
you saw when you removed the line. With a bad record first, nothing would ever
be inserted, and the test would pass with or without a transaction.
What you can now do
Put storage in its own module whose functions take the connection as a
parameter.
Give every test its own :memory: database with setUp and addCleanup.
Make an import idempotent with a primary key and ON CONFLICT … DO NOTHING.
Rely on with conn: for all-or-nothing saves, and prove it with a test.
Choose test inputs that could actually expose the bug, like unsorted data or a
failure in the middle.
Next: turning a summary into JSON that another program can read, without
losing anything on the way.
Your progress is saved in this browser.
Sign in
to keep it on every device.