The analyzer’s report is for people. But the numbers are also wanted by
programs: a dashboard that draws the error rate, or an alert that fires when
payments errors spike. Those programs shouldn’t have to pick numbers out of a
printout. They need data, and on the web and between programs, data usually
means JSON.
You met the json module in the
beginner course. Today is about what happens at
its edges.
What JSON can hold
JSON has six kinds of value, and each maps onto a Python type:
JSON
Python
object {"a": 1}
dict
array [1, 2]
list (a tuple is written as an array too)
string "text"
str
number 1, 2.5
int, float
true / false
True / False
null
None
Anything else, like a datetime, a set or your own class, has no JSON
equivalent, and json refuses it:
What json writes, and what it refusesPython
import jsonfrom collections import Counterfrom datetime import datetimesummary = { "records": 6, "levels": Counter({"ERROR": 3, "INFO": 2, "WARNING": 1}), # a dict subclass: fine "sources": ("auth", "payments"), # a tuple: becomes a list "complete": True, "notes": None,}print(json.dumps(summary, indent=2))try: json.dumps({"first": datetime(2026, 9, 1, 9, 14, 2)})except TypeError as error: print("TypeError:", error)
The refusal is the helpful case. You find out straight away, and you convert
the value yourself, usually a datetime to ISO text with .isoformat(), just
as you did for SQLite. The quiet changes are the dangerous ones. The tuple came
back as a list, and nothing said so.
Bug hunt: the numbers that stopped matching
The summary gains an errors-per-hour count, a dict from hour to count. A test
checks the summary survives being saved and loaded:
Save, load, comparePython
import jsonimport sysimport unittestfrom collections import Counterdef errors_per_hour(hours): """How many errors started in each hour of the day.""" return dict(Counter(hours))def save_summary(summary, path): with open(path, "w") as file: json.dump(summary, file, indent=2)def load_summary(path): with open(path) as file: return json.load(file)class TestSummaryFile(unittest.TestCase): def test_summary_survives_a_round_trip(self): summary = {"records": 6, "errors_per_hour": errors_per_hour([9, 9, 10])} save_summary(summary, "summary.json") self.assertEqual(load_summary("summary.json"), summary)suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestSummaryFile)unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Reproduce. One small summary, and the round trip fails.
Inspect. Compare the two sides of the failure: {'9': 2, '10': 1} against
{9: 2, 10: 1}. The counts are right. The keys changed type. To see where,
look at the file itself. Add this at the end of the playground:
print(open("summary.json").read())
In the file the keys are already "9" and "10", in quotes.
Before you open this, make the round trip pass without changing the test. There’s more than one way. Then compare with ours, which is about choosing the file’s shape, not just converting values.
Show the diagnosis and the fix
Narrow. JSON object keys can only be strings. json.dump quietly turns
integer keys into their text, and json.load has no way of knowing that "9"
used to be a number. (A few other key types are converted too, like True to
"true". Keys it can’t convert at all, such as tuples, raise TypeError.) Anyone reading this file gets strings, in every language.
A dashboard that looks up data["errors_per_hour"][9] gets a KeyError.
Pin it. The round-trip test is the regression test, and it’s worth having
for any data you write to a file: what goes out must come back equal.
Fix. You could convert the keys back after loading, but then every program
that reads the file has to know to do that. Change the shape instead, so that
nothing needs converting. Use a list of small objects, one per hour:
def errors_per_hour(hours): """How many errors started in each hour of the day, in hour order.""" return [{"hour": hour, "errors": count} for hour, count in sorted(Counter(hours).items())]
The hour is a real number now, the order is explicit rather than an accident
of insertion, and each entry says what its fields are. It’s more verbose, and
it’s the shape most JSON APIs use for exactly this reason.
Design the file for its readers
A JSON file is an interface, like a function’s signature, and other programs
will depend on it. Keep keys as fixed names chosen by you ("hour", "errors"),
not data ("9", "payments"). Put data in values, where it keeps its type. A
dict keyed by source name is fine inside your program. In a file, a list of
{"source": …, "count": …} objects ages better.
The Day 3 summary
store.py builds the whole summary straight from the database, in the shape
above. SQL does the counting, and substr(timestamp, 12, 2) picks the hour out
of the ISO text:
def summary(conn): """Everything the daily report needs, as plain JSON-ready values.""" total, first, last = conn.execute( "SELECT COUNT(*), MIN(timestamp), MAX(timestamp) FROM records").fetchone() hours = conn.execute( "SELECT CAST(substr(timestamp, 12, 2) AS INTEGER) AS hour, COUNT(*) AS n " "FROM records WHERE level = 'ERROR' GROUP BY hour ORDER BY hour") return { "records": total, "first": first, "last": last, "levels": count_by(conn, "level"), "errors_by_source": count_by(conn, "source", level="ERROR"), "errors_by_hour": [{"hour": row["hour"], "errors": row["n"]} for row in hours], }
first and last come out of the database as ISO text already, so they go
straight into JSON. levels and errors_by_source stay dicts: their keys are
names, which are strings already, so they survive the round trip unchanged. By
the callout’s advice they could be lists of objects too. Keeping them as dicts
is a judgement call, made on purpose because every reader so far is a small
script. That’s the difference between a choice and an accident.
Day 3 checkpoint
Exercise
Store and summarise web requests
Back to Day 2’s access log. requests_store.py has the Request record, the
table and connect(). Write the four functions so the tests pass:
save_requests(conn, requests) stores requests, is safe to run twice, and
returns how many were new. Timestamps go in as ISO text.
status_counts(conn) returns {status: count} in status order.
error_paths(conn, min_status=400) returns the distinct paths that got a
status of at least min_status, sorted.
export_summary(conn, path) writes the JSON the last test describes.
One of the paths contains an apostrophe, which you’ll only get past with
placeholders. And one of the results is keyed by an integer, which you’ll need
to reshape before it goes into JSON.
"""Day 3 checkpoint: store access-log requests and report on them. The tests are the spec."""
import json
import sqlite3
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class Request:
ip: str
timestamp: datetime
method: str
path: str
status: int
size: int
line_number: int = 0
SCHEMA = """
CREATE TABLE IF NOT EXISTS requests (
line_number INTEGER PRIMARY KEY,
ip TEXT NOT NULL,
timestamp TEXT NOT NULL,
method TEXT NOT NULL,
path TEXT NOT NULL,
status INTEGER NOT NULL,
size INTEGER NOT NULL
)
"""
def connect(path=":memory:"):
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute(SCHEMA)
return conn
def save_requests(conn, requests):
"""Store requests, skipping line numbers already stored. Return how many were new."""
raise NotImplementedError("your code here")
def status_counts(conn):
"""{status: number of requests}, in status order."""
raise NotImplementedError("your code here")
def error_paths(conn, min_status=400):
"""Distinct paths that got a status of min_status or above, sorted."""
raise NotImplementedError("your code here")
def export_summary(conn, path):
"""Write the summary the tests describe as JSON to path."""
raise NotImplementedError("your code here")
import json
import unittest
from datetime import datetime, timezone
from requests_store import (Request, connect, error_paths, export_summary,
save_requests, status_counts)
WHEN = datetime(2026, 9, 1, 9, 14, 2, tzinfo=timezone.utc)
REQUESTS = [
Request("10.0.0.5", WHEN, "GET", "/books", 200, 5120, 1),
Request("10.0.0.6", WHEN, "GET", "/books", 200, 4800, 2),
Request("10.0.0.5", WHEN, "GET", "/missing", 404, 0, 3),
Request("10.0.0.7", WHEN, "POST", "/checkout", 500, 0, 4),
Request("10.0.0.7", WHEN, "GET", "/o'reilly", 404, 0, 5),
]
class TestRequestsStore(unittest.TestCase):
def setUp(self):
self.conn = connect()
self.addCleanup(self.conn.close)
def test_saving_twice_stores_once(self):
self.assertEqual(save_requests(self.conn, REQUESTS), 5)
self.assertEqual(save_requests(self.conn, REQUESTS), 0)
def test_timestamps_are_stored_as_iso_text(self):
save_requests(self.conn, REQUESTS[:1])
stored = self.conn.execute("SELECT timestamp FROM requests").fetchone()[0]
self.assertEqual(stored, "2026-09-01T09:14:02+00:00")
def test_status_counts(self):
save_requests(self.conn, REQUESTS)
self.assertEqual(status_counts(self.conn), {200: 2, 404: 2, 500: 1})
def test_error_paths(self):
save_requests(self.conn, REQUESTS)
self.assertEqual(error_paths(self.conn), ["/checkout", "/missing", "/o'reilly"])
self.assertEqual(error_paths(self.conn, min_status=500), ["/checkout"])
def test_export_summary(self):
save_requests(self.conn, REQUESTS)
export_summary(self.conn, "requests.json")
with open("requests.json") as file:
data = json.load(file)
self.assertEqual(data, {
"requests": 5,
"statuses": [{"status": 200, "count": 2},
{"status": 404, "count": 2},
{"status": 500, "count": 1}],
"error_paths": ["/checkout", "/missing", "/o'reilly"],
})
import sys
import unittest
import test_requests_store
suite = unittest.defaultTestLoader.loadTestsFromModule(test_requests_store)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Hint 1: the idea
Each function is one of today’s lessons in a new setting. save_requests is
the idempotent insert from lesson 3’s bug hunt: a conflict on the key does
nothing, and the cursor reports how many rows really went in. The two queries
are lesson 2’s placeholders with GROUP BY and DISTINCT. export_summary is
this lesson’s reshaping: JSON keys must be strings, so a dictionary keyed by
status codes becomes a list of small objects.
Hint 2: a first step
Work in test order, running after each function. For save_requests, turn each
Request into a tuple — timestamp.isoformat() for the time — and hand them all
to executemany inside with conn:. Read what the test expects it to return
the second time before deciding how to compute the count.
Show one solution
def save_requests(conn, requests): """Store requests, skipping line numbers already stored. Return how many were new.""" rows = ((r.line_number, r.ip, r.timestamp.isoformat(), r.method, r.path, r.status, r.size) for r in requests) with conn: cursor = conn.executemany( "INSERT INTO requests VALUES (?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT (line_number) DO NOTHING", rows) return cursor.rowcountdef status_counts(conn): """{status: number of requests}, in status order.""" rows = conn.execute( "SELECT status, COUNT(*) AS n FROM requests GROUP BY status ORDER BY status") return {row["status"]: row["n"] for row in rows}def error_paths(conn, min_status=400): """Distinct paths that got a status of min_status or above, sorted.""" rows = conn.execute( "SELECT DISTINCT path FROM requests WHERE status >= ? ORDER BY path", (min_status,)) return [row["path"] for row in rows]def export_summary(conn, path): """Write the summary the tests describe as JSON to path.""" total = conn.execute("SELECT COUNT(*) FROM requests").fetchone()[0] data = { "requests": total, "statuses": [{"status": status, "count": count} for status, count in status_counts(conn).items()], "error_paths": error_paths(conn), } with open(path, "w") as file: json.dump(data, file, indent=2)
Each function uses one of today’s lessons:
save_requests converts the timestamp at the edge, saves in one
transaction, and is idempotent through the line_number primary key.
status_counts is fine as a dict inside Python, where 200 stays an
int. It’s export_summary that reshapes it into a list of objects, at the
point where it leaves for JSON.
error_paths passes min_status as a ? placeholder. The apostrophe in
/o'reilly is only ever data, so it can’t break anything.
export_summary reuses the other two functions instead of repeating their
SQL. The JSON is built from the same answers the rest of the program sees.
The Day 3 reference solution
The analyzer at the end of Day 3, and where Day 4 starts. main.py parses
messy.log, stores the records in an in-memory database, prints a count
straight from SQL, writes summary.json and shows it, then runs all nineteen
tests.
Day 3 reference: parse, store, query, exportPython"""The Day 3 report: parse the log, store it, then ask the database."""
import sys
import unittest
import test_analyzer
import test_store
from analyzer import parse_lines, read_lines
from store import connect, count_by, save_records, save_summary, summary
conn = connect() # in memory; connect("logs.db") would keep it in a file
errors = []
stored = save_records(conn, "messy.log", parse_lines(read_lines("messy.log"), errors))
print(f"Stored {stored} records; skipped {len(errors)} bad line(s).")
print("Errors by source:", count_by(conn, "source", level="ERROR"))
save_summary(summary(conn), "summary.json")
with open("summary.json") as file:
print(file.read())
print()
loader = unittest.defaultTestLoader
suite = unittest.TestSuite([loader.loadTestsFromModule(test_analyzer),
loader.loadTestsFromModule(test_store)])
unittest.TextTestRunner(stream=sys.stdout, verbosity=1).run(suite)
"""Log analyzer, Day 3: keep parsed records in SQLite, and ask questions in SQL."""
import json
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 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}%"
def search(conn, text):
"""Records whose message contains text (ignoring ASCII case), oldest first."""
rows = conn.execute(
"SELECT * FROM records WHERE message LIKE ? ESCAPE '\\' ORDER BY timestamp",
(like_pattern(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}
def summary(conn):
"""Everything the daily report needs, as plain JSON-ready values."""
total, first, last = conn.execute(
"SELECT COUNT(*), MIN(timestamp), MAX(timestamp) FROM records").fetchone()
hours = conn.execute(
"SELECT CAST(substr(timestamp, 12, 2) AS INTEGER) AS hour, COUNT(*) AS n "
"FROM records WHERE level = 'ERROR' GROUP BY hour ORDER BY hour")
return {
"records": total,
"first": first,
"last": last,
"levels": count_by(conn, "level"),
"errors_by_source": count_by(conn, "source", level="ERROR"),
"errors_by_hour": [{"hour": row["hour"], "errors": row["n"]} for row in hours],
}
def save_summary(data, path):
with open(path, "w") as file:
json.dump(data, file, indent=2)
def load_summary(path):
with open(path) as file:
return json.load(file)
"""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 sqlite3
import unittest
from dataclasses import replace
from analyzer import ParseError, parse_record
from store import (connect, count_by, load_records, load_summary, save_records,
save_summary, search, summary)
LINES = [
"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\"",
"2026-09-01 10:02:11 ERROR db: msg=\"can't reach replica\"",
]
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_records_come_back_equal(self): # bug hunt 1
save_records(self.conn, "a.log", records())
self.assertEqual(load_records(self.conn, "a.log"), records())
def test_search_text_with_a_quote(self): # bug hunt 2
save_records(self.conn, "a.log", records())
found = search(self.conn, "can't")
self.assertEqual([r.source for r in found], ["db"])
def test_search_is_not_sql(self):
save_records(self.conn, "a.log", records())
self.assertEqual(search(self.conn, "x' OR '1'='1"), [])
def test_saving_the_same_log_twice_stores_it_once(self): # bug hunt 3
self.assertEqual(save_records(self.conn, "a.log", records()), 3)
self.assertEqual(save_records(self.conn, "a.log", records()), 0)
self.assertEqual(count_by(self.conn, "level"), {"ERROR": 2, "INFO": 1})
def test_search_results_are_oldest_first(self): # Day 3 exercise
save_records(self.conn, "a.log", reversed(records())) # stored newest first
self.assertEqual([r.source for r in search(self.conn, "msg=")], ["payments", "db"])
def test_search_treats_wildcards_as_text(self):
save_records(self.conn, "a.log", records())
self.assertEqual(search(self.conn, "%"), [])
self.assertEqual(search(self.conn, "_"), [])
def test_a_failed_save_stores_nothing(self): # Day 3 exercise
good = records()
bad = replace(good[1], source=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"), {})
def test_a_stream_that_fails_part_way_stores_nothing(self):
def stream():
yield records()[0]
raise ParseError(2, "too few parts", "garbage") # fail-fast parsing, mid-save
with self.assertRaises(ParseError):
save_records(self.conn, "a.log", stream())
self.assertEqual(count_by(self.conn, "level"), {})
def test_count_by_one_level(self):
save_records(self.conn, "a.log", records())
self.assertEqual(count_by(self.conn, "source", level="ERROR"),
{"db": 1, "payments": 1})
def test_count_by_refuses_other_columns(self):
with self.assertRaises(ValueError):
count_by(self.conn, "message; DROP TABLE records")
class TestSummary(unittest.TestCase):
def test_summary_survives_a_round_trip_through_json(self): # bug hunt 4
conn = connect()
self.addCleanup(conn.close)
save_records(conn, "a.log", records())
data = summary(conn)
save_summary(data, "test-summary.json")
self.assertEqual(load_summary("test-summary.json"), data)
self.assertEqual(data["errors_by_hour"], [{"hour": 9, "errors": 1},
{"hour": 10, "errors": 1}])
import unittest
from datetime import datetime
from analyzer import (ParseError, parse_fields, parse_lines, parse_record,
summarize)
class TestParseRecord(unittest.TestCase):
def test_prefix_becomes_fields(self):
record = parse_record("2026-09-01 09:14:02 INFO auth: user=ada action=login")
self.assertEqual(record.timestamp, datetime(2026, 9, 1, 9, 14, 2))
self.assertEqual((record.level, record.source), ("INFO", "auth"))
def test_message_has_no_leading_space(self): # bug hunt 1
record = parse_record("2026-09-01 09:14:02 INFO auth: user=ada")
self.assertEqual(record.message, "user=ada")
class TestParseFields(unittest.TestCase):
def test_quoted_value_keeps_its_spaces(self):
self.assertEqual(parse_fields('msg="card declined"'), {"msg": "card declined"})
def test_two_quoted_values_stay_separate(self): # bug hunt 2
fields = parse_fields('query="python books" msg="index offline"')
self.assertEqual(fields, {"query": "python books", "msg": "index offline"})
class TestStreaming(unittest.TestCase):
def test_summary_works_on_a_generator(self): # bug hunt 3
lines = ["2026-09-01 09:00:00 ERROR db: x=1",
"2026-09-01 09:00:01 INFO db: x=2"]
levels, errors = summarize(parse_lines(iter(lines), []))
self.assertEqual(levels, {"ERROR": 1, "INFO": 1})
self.assertEqual(errors, {"db": 1})
class TestBadInput(unittest.TestCase):
def test_each_kind_of_bad_line_is_reported(self):
cases = {
"garbage": "too few parts",
"2026-13-01 09:00:00 INFO a: x=1": "bad timestamp",
"2026-09-01 09:00:00 LOUD a: x=1": "unknown level 'LOUD'",
"2026-09-01 09:00:00 INFO no colon here": "missing 'source:'",
}
for line, reason in cases.items():
with self.subTest(line=line):
with self.assertRaises(ParseError) as caught:
parse_record(line, 7)
self.assertEqual(caught.exception.reason, reason)
self.assertEqual(caught.exception.line_number, 7)
def test_bad_lines_are_collected_not_fatal(self):
errors = []
records = list(parse_lines(["garbage", "", "2026-09-01 09:00:00 INFO a: x=1"], errors))
self.assertEqual(len(records), 1)
self.assertEqual([e.line_number for e in errors], [1])
def test_every_line_is_accounted_for(self): # bug hunt 4
lines = ["garbage", "", "2026-09-01 09:00:00 INFO a: x=1", " "]
errors = []
records = list(parse_lines(lines, errors))
non_blank = [line for line in lines if line.strip()]
self.assertEqual(len(records) + len(errors), len(non_blank))
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"
Working on your own machine? Download the Day 3 files
and carry on there. You need Python 3.12 or later, and nothing else.
Going further: teaching json new types
Optional
Nothing later in the course depends on this section.
If you write the same kinds of value to JSON in many places, you can teach
json to convert them. Subclass JSONEncoder and override default, which
json calls for any value it doesn’t know:
A custom encoderPython
import jsonfrom datetime import datetimeclass SummaryEncoder(json.JSONEncoder): """Also knows how to write datetimes and sets.""" def default(self, value): if isinstance(value, datetime): return value.isoformat() if isinstance(value, set): return sorted(value) return super().default(value) # anything else: the usual TypeErrordata = {"first": datetime(2026, 9, 1, 9, 14, 2), "sources": {"search", "auth"}}print(json.dumps(data, cls=SummaryEncoder))print(json.dumps(data, default=str)) # the quick version: str() on anything unknown
Compare the two lines of output. default=str is tempting because it turns
almost anything into something, and that’s the problem: the set comes out as one string that looks like
Python, which no other program can use. The encoder converts only the types it
knows, and still refuses everything else. The same rule as allow-lists in SQL:
say what’s allowed, and reject the rest.
Going the other way is harder. JSON has no way to mark “this string is a date”,
so turning ISO strings back into datetimes is up to the reader, which is one
more reason to keep file formats simple.
What you can now do
Predict which Python values JSON keeps, changes, or refuses.
Test that saved data comes back equal, and find what didn’t.
Design a JSON shape with fixed keys and typed values, rather than data as
keys.
Build a report from SQL and export it for other programs.
Tomorrow the analyzer becomes configurable: which lines to keep and how to
group them become arguments, and those arguments are functions.
Your progress is saved in this browser.
Sign in
to keep it on every device.