Rejecting Bad Input

Real log files are not tidy. Log rotation writes a marker line, one service logs its times without seconds, and a crash leaves half a line behind. Here is what Day 1’s level_of makes of three of them:

def level_of(line):
    """Day 1: the level is the third word of a line."""
    return line.split()[2]


junk = [
    "--- rotated at 09:16 ---",
    "2026-09-01 9:18 INFO search: query=dune results=4",
    "garbage",
]
for line in junk:
    print(repr(level_of(line)))

The crash on the last line isn’t the worst part. The first two lines didn’t crash at all. The rotation marker has the level 'at', and the second line comes out as 'INFO', which looks right but only by accident: the time is malformed, and nothing noticed. A program that crashes on bad input is annoying. A program that quietly accepts it is wrong.

Decide what a bad line means

Before writing any code, decide what the program should do with a bad line. There are two reasonable answers:

A log analyzer should skip and report. Whichever you choose, one thing is never acceptable: skipping without reporting. A count that silently leaves out 5% of the lines is worse than no count, because someone will believe it.

Both strategies start from the same place. parse_record checks the line and raises a precise error when it’s wrong. The caller decides whether that error stops the program or gets collected.

An exception that says what went wrong

ValueError("bad line") would work, but the caller can’t do much with it. An error for bad input should carry what the caller needs: which line, why it was rejected, and the line itself.

from datetime import datetime

LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}


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 check_prefix(line, line_number=0):
    parts = line.split(maxsplit=3)
    if len(parts) < 4:
        raise ParseError(line_number, "too few parts", line)
    date, time, level, rest = parts
    try:
        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)
    if ":" not in rest:
        raise ParseError(line_number, "missing 'source:'", line)
    return "ok"


lines = [
    "2026-09-01 09:14:02 INFO    auth: user=ada",
    "garbage",
    "2026-09-01 9:18 INFO search: query=dune",
    "2026-09-01 09:20:00 LOUD    auth: user=ada",
    "2026-09-01 09:21:00 INFO    no colon anywhere",
]
for number, line in enumerate(lines, start=1):
    try:
        print(check_prefix(line, number))
    except ParseError as error:
        print(error, "| reason:", error.reason)

Four design choices are worth copying:

• It subclasses ValueError. “The value you gave me is wrong” is exactly what ValueError means, so code that already catches ValueError keeps working. Code that wants only parse errors can catch ParseError.
• Its data are attributes, not just text. The message is for people. reason and line_number are for code, like a test that checks why a line was rejected, or a report that groups bad lines by reason.
• It checks in order, cheapest first. Each check can then assume the ones before it passed. There’s no point parsing a timestamp from a line with two words.
• from None hides the internal error. Without it, the traceback would show fromisoformat’s own ValueError first and then “During handling of the above exception, another exception occurred”. That’s noise here, because ParseError already says everything. Use raise ... from error instead when the original cause is useful to the reader.

Testing the bad cases

Each kind of bad line needs a test, and they’re all the same test with different data. subTest runs one test body over a table of cases and reports each failure on its own, so one broken case doesn’t hide the rest:

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)

assertRaises used as a with block catches the exception, and caught.exception lets you check its attributes afterwards. Checking only that some ParseError was raised would pass even if the line was rejected for the wrong reason.

Bug hunt: the numbers that looked fine

The analyzer’s parse_lines collects bad lines into a list the caller passes in. Someone wrote the report below, and it looks perfectly healthy. Run it, then look in messy.log:

Reproduce. The report says it counted 6 records. messy.log has 8 non-blank lines. Nothing crashed, and nothing printed a warning.

Inspect. Where did two lines go? parse_lines doesn’t raise for a bad line. It appends a ParseError to errors and moves on. So the relevant state is the errors list, and main.py passes []: a brand-new list that nothing else can see. The errors were collected, then thrown away with the list.

Narrow. Give the list a name and print it:

errors = []
levels, error_sources = summarize(parse_lines(read_lines("messy.log"), errors))
print(errors)

There they are: the rotation marker, and the line with the short time.

Pin it. Write a test for the rule itself rather than for this one file: every non-blank line becomes either a record or an error.

def test_every_line_is_accounted_for(self):
    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))

Fix. Keep the list and report it. That’s the Day 2 reference below.

Two tools for the checkpoint

The checkpoint parses a different kind of log, and it needs two tools the week hasn’t used yet. Try them here first.

Matching a whole line. re has three ways to apply a pattern, and they differ in how much of the text the pattern must cover:

import re

STATUS = re.compile(r"(?P<code>\d{3}) (?P<size>\d+)")

for text in ["200 5120", "200 5120 and then junk", "junk, then 200 5120"]:
    print(f"{text!r:28}",
          "search:", bool(STATUS.search(text)),       # anywhere in the text
          "match:", bool(STATUS.match(text)),         # starting at the beginning
          "fullmatch:", bool(STATUS.fullmatch(text))) # the whole text, nothing else

When a line must be a record, with nothing before or after it, use fullmatch. It returns None for anything else, which is exactly the signal for a ParseError.

Reading a timestamp in someone else’s format. datetime.fromisoformat only reads ISO dates. For any other layout, datetime.strptime(text, format) takes a format string that spells the layout out, one directive per part:

from datetime import datetime

when = datetime.strptime("01/Sep/2026:09:14:02 +0000", "%d/%b/%Y:%H:%M:%S %z")
print(when)
print(when.tzinfo)                     # %z read "+0000": this time knows its timezone

for bad in ["31/Feb/2026:09:14:02 +0000", "01/Sep/2026 09:14:02"]:
    try:
        datetime.strptime(bad, "%d/%b/%Y:%H:%M:%S %z")
    except ValueError as error:
        print("ValueError:", error)

%d is the day, %b the short month name, %Y the four-digit year, %H:%M:%S the time, and %z a UTC offset like +0000. Everything else in the format, like / and :, must appear literally. A regex can check that a date looks right. Only strptime knows that February has no 31st.

Day 2 checkpoint

10.0.0.5 - - [01/Sep/2026:09:14:02 +0000] "GET /books?q=python HTTP/1.1" 200 5120

ACCESS = re.compile(
    r"""
    (?P<ip>\S+) \s \S+ \s \S+ \s          # 10.0.0.5 - -
    \[(?P<when>[^\]]+)\] \s               # [01/Sep/2026:09:14:02 +0000]
    "(?P<method>[A-Z]+) \s (?P<path>\S+) \s [^"]*" \s   # "GET /books HTTP/1.1"
    (?P<status>\d{3}) \s (?P<size>\d+|-)  # 200 5120
    """,
    re.VERBOSE,
)


def parse_access(line, line_number=0):
    match = ACCESS.fullmatch(line)
    if match is None:
        raise ParseError(line_number, "not an access log line", line)
    try:
        timestamp = datetime.strptime(match["when"], "%d/%b/%Y:%H:%M:%S %z")
    except ValueError:
        raise ParseError(line_number, "bad timestamp", line) from None
    size = 0 if match["size"] == "-" else int(match["size"])
    return Request(match["ip"], timestamp, match["method"], match["path"],
                   int(match["status"]), size, line_number)


def parse_access_lines(lines, errors):
    for number, line in enumerate(lines, start=1):
        if not line.strip():
            continue
        try:
            yield parse_access(line, number)
        except ParseError as error:
            errors.append(error)


def busiest_paths(requests, n):
    """The n most requested paths, ignoring query strings."""
    counts = Counter(request.path.partition("?")[0] for request in requests)
    return counts.most_common(n)

The Day 2 reference solution

This is the analyzer at the end of the day, and where Day 3 starts. Run prints the report from messy.log, including the two lines it skipped, then runs all eight tests. Four of them are the regression tests from today’s bug hunts.

Working on your own machine? Download the Day 2 files and carry on there. You need Python 3.12 or later, and nothing else.

If your own version differs, that’s fine. What matters is that it passes the same tests. Compare it with the reference for anything you’d like to borrow.

Going further: matching on records

Once lines are records, match can pick them apart by shape. A class pattern checks the type and pulls out attributes in one step, and a guard (if …) adds a condition:

from dataclasses import dataclass


@dataclass(frozen=True)
class LogRecord:
    level: str
    source: str
    fields: dict


def describe(record):
    match record:
        case LogRecord(level="ERROR", source="payments", fields={"order": order}):
            return f"payment failed for order {order}"
        case LogRecord(level="ERROR", source=source):
            return f"error in {source}"
        case LogRecord(fields={"retry": retry}) if int(retry) > 2:
            return f"retried {retry} times"
        case _:
            return "nothing to report"


print(describe(LogRecord("ERROR", "payments", {"order": "1043", "msg": "declined"})))
print(describe(LogRecord("ERROR", "search", {})))
print(describe(LogRecord("WARNING", "auth", {"retry": "5"})))
print(describe(LogRecord("INFO", "auth", {"user": "ada"})))

A mapping pattern like {"order": order} matches any dict that has that key, whatever else it holds. Cases are tried top to bottom, so the specific payment case has to come before the general error case. That’s the same “most specific first” rule as regex alternation.

What you can now do

• Choose between failing fast and skipping-and-reporting, and never skip without reporting.
• Design an exception that carries what its caller needs: a reason, a location and the bad input.
• Test every way input can be wrong with subTest and assertRaises, checking why it was rejected, not just that it was.
• Write a test for an invariant (“every line is accounted for”) instead of for one example.

Tomorrow the records stop disappearing when the program ends: they go into a database.