Unit 8 Project: Parse a Log File into a Database
Messy text in, queryable table out
In this lesson
This project uses both halves of the unit at once, which is why they are paired. You will read a messy log line by line, pull four fields out of each with a single pattern, store them in a table, and then ask the questions that were impossible while it was still text: which error happens most, and when.
Explain it like I’m 5
You are turning a pile of shouting into a tidy spreadsheet you can ask questions about.
Read the log without loading it all
def read_lines(path):
"""Yield one line at a time, so the file size stops mattering."""
with open(path, encoding="utf-8") as handle:
for line in handle:
yield line.rstrip("\n")
# for line in read_lines("app.log"):
# ...
One pattern, four fields
Each well-formed line has the same shape, so one regular expression with four capture groups turns it into a tuple ready for a database row. Compile the pattern once outside the loop with re.compile(), since it is used on every line.
Real logs contain lines that do not fit: blank lines, stack traces, output from something else. Those must not crash the parser. Count them and move on, then report the number at the end so a broken pattern is visible rather than silent.
import re
PATTERN = re.compile(
r"^(\d{4}-\d{2}-\d{2}) (\d{2}):\d{2}:\d{2} (ERROR|INFO|WARN) (.*)$"
)
line = "2026-07-27 10:02:11 ERROR E404 missing file"
match = PATTERN.match(line)
print(match.groups())
('2026-07-27', '10', 'ERROR', 'E404 missing file')
Store it, then ask questions
With the fields extracted, the rest is Unit 8’s first half: create a table, insert with placeholders, commit once at the end rather than inside the loop, and query.
Committing once matters. A commit() per row on a large file is dramatically slower, because each one is a separate transaction. Build up the inserts, then commit the batch.
counts = conn.execute("""
SELECT message, COUNT(*) AS total
FROM events
WHERE level = ?
GROUP BY message
ORDER BY total DESC
LIMIT 10
""", ("ERROR",)).fetchall()
for row in counts:
print(row["total"], row["message"])
Build the parser. Loop over the log lines, skip any that don’t match (counting them), insert the rest, then print the skipped count and a count of events per level, ordered by level.
import re, sqlite3
LOG = """2026-07-27 09:14:02 ERROR disk full
-- not a log line --
2026-07-27 09:15:31 INFO startup complete
2026-07-27 10:02:11 ERROR missing file
2026-07-27 10:30:00 ERROR disk full"""
PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}) (\d{2}):\d{2}:\d{2} (ERROR|INFO) (.*)$")
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE events (day TEXT, hour TEXT, level TEXT, message TEXT)")
skipped = 0
for line in LOG.splitlines():
pass # TODO: match; count skips; insert match.groups()
conn.commit()
print("skipped:", skipped)
for row in conn.execute("SELECT level, COUNT(*) FROM events GROUP BY level ORDER BY level"):
print(row[0], row[1])Inside the loop: match = PATTERN.match(line), then if match is None: increment skipped and continue. Otherwise pass match.groups() straight in as the four placeholder values.
import re, sqlite3
LOG = """2026-07-27 09:14:02 ERROR disk full
-- not a log line --
2026-07-27 09:15:31 INFO startup complete
2026-07-27 10:02:11 ERROR missing file
2026-07-27 10:30:00 ERROR disk full"""
PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}) (\d{2}):\d{2}:\d{2} (ERROR|INFO) (.*)$")
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE events (day TEXT, hour TEXT, level TEXT, message TEXT)")
skipped = 0
for line in LOG.splitlines():
match = PATTERN.match(line)
if match is None:
skipped += 1
continue
conn.execute("INSERT INTO events VALUES (?, ?, ?, ?)", match.groups())
conn.commit()
print("skipped:", skipped)
for row in conn.execute("SELECT level, COUNT(*) FROM events GROUP BY level ORDER BY level"):
print(row[0], row[1])Common mistake: Crashing on the first line that doesn’t match
match.groups() on a None raises AttributeError, and sample logs are usually tidy.
Check if match is None before using it. Count the skips and print the total, so a pattern that quietly matches nothing is obvious.
Common mistake: Calling commit() inside the loop
Committing each row feels safer than waiting until the end.
Each commit is a separate transaction, so per-row commits are far slower on a large file. Insert everything, then commit once.
Common mistake: Storing the whole raw line instead of the fields
It is less work up front and keeps all the information.
Then every query has to re-parse the text and you have gained nothing over the file. Store the extracted fields in their own columns, which is the entire point.
Common mistake: Reading the whole file with .read()
It works instantly on a small sample.
Loop over the file object instead. It uses a constant amount of memory whatever the file size, and the code is shorter.
What should happen to a log line the pattern does not match?
Real logs always contain odd lines. Counting skips keeps the run going and makes a broken pattern visible.
Why commit once at the end rather than after each insert?
Batching the inserts into one transaction is dramatically faster on a large file.
Which query answers 'which error message is most common'?
GROUP BY collapses identical messages and COUNT(*) counts each group; ordering descending puts the worst first.
Mini exercise (hard)
Extend the parser. After loading the events, write busiest_hour(conn) returning the (hour, count) with the most ERROR events, and top_errors(conn, n) returning the n most common error messages as (message, count) tuples.
Take the wheel. Complete the code, hit Run, and check your output right here.
import re, sqlite3
LOG = """2026-07-27 09:14:02 ERROR disk full
2026-07-27 09:15:31 INFO startup complete
2026-07-27 10:02:11 ERROR missing file
2026-07-27 10:30:00 ERROR disk full"""
PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}) (\d{2}):\d{2}:\d{2} (ERROR|INFO) (.*)$")
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE events (day TEXT, hour TEXT, level TEXT, message TEXT)")
for line in LOG.splitlines():
match = PATTERN.match(line)
if match:
conn.execute("INSERT INTO events VALUES (?, ?, ?, ?)", match.groups())
conn.commit()
def busiest_hour(conn):
return None # TODO: (hour, count) with the most ERROR events
def top_errors(conn, n):
return [] # TODO: the n most common error messages as (message, count)
print(busiest_hour(conn))
print(top_errors(conn, 2))Both are GROUP BY queries with WHERE level = ?. Group by hour for the first and by message for the second, then ORDER BY COUNT(*) DESC LIMIT ?.
import re, sqlite3
LOG = """2026-07-27 09:14:02 ERROR disk full
2026-07-27 09:15:31 INFO startup complete
2026-07-27 10:02:11 ERROR missing file
2026-07-27 10:30:00 ERROR disk full"""
PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}) (\d{2}):\d{2}:\d{2} (ERROR|INFO) (.*)$")
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE events (day TEXT, hour TEXT, level TEXT, message TEXT)")
for line in LOG.splitlines():
match = PATTERN.match(line)
if match:
conn.execute("INSERT INTO events VALUES (?, ?, ?, ?)", match.groups())
conn.commit()
def busiest_hour(conn):
row = conn.execute("""
SELECT hour, COUNT(*) AS total FROM events
WHERE level = ? GROUP BY hour ORDER BY total DESC LIMIT 1
""", ("ERROR",)).fetchone()
return (row["hour"], row["total"])
def top_errors(conn, n):
rows = conn.execute("""
SELECT message, COUNT(*) AS total FROM events
WHERE level = ? GROUP BY message ORDER BY total DESC, message LIMIT ?
""", ("ERROR", n)).fetchall()
return [(row["message"], row["total"]) for row in rows]
print(busiest_hour(conn))
print(top_errors(conn, 2))