python 39 lines · 6 steps

Merging sorted log files with heapq

Stream multiple time-sorted log files and interleave them into one chronological output without loading everything into memory.

Explained by highlit
Intermediate generators heapq lazy-evaluation merge namedtuple
1import heapq
2from datetime import datetime
3from pathlib import Path
4from typing import Iterator, NamedTuple
5 
6 
7class LogEntry(NamedTuple):
8 timestamp: datetime
9 source: str
10 message: str
11 
12 
13def parse_entries(path: Path) -> Iterator[LogEntry]:
14 source = path.stem
15 with path.open(encoding="utf-8") as fh:
16 for line in fh:
17 line = line.rstrip("\n")
18 if not line:
19 continue
20 stamp, _, message = line.partition(" ")
21 try:
22 ts = datetime.fromisoformat(stamp)
23 except ValueError:
24 continue
25 yield LogEntry(ts, source, message)
26 
27 
28def merge_logs(paths: list[Path]) -> Iterator[LogEntry]:
29 streams = [parse_entries(p) for p in paths]
30 yield from heapq.merge(*streams, key=lambda entry: entry.timestamp)
31 
32 
33def write_merged(paths: list[Path], destination: Path) -> int:
34 count = 0
35 with destination.open("w", encoding="utf-8") as out:
36 for entry in merge_logs(paths):
37 out.write(f"{entry.timestamp.isoformat()} [{entry.source}] {entry.message}\n")
38 count += 1
39 return count
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1heapq.merge interleaves already-sorted iterables lazily, so you never hold every entry in memory at once.
  2. 2Generators with yield let each file be read line-by-line on demand, keeping the whole pipeline streaming.
  3. 3A NamedTuple gives structured, self-documenting records while staying lightweight and comparable.

Related explainers

python 
import click
from flask.cli import AppGroup
 
from . import db

Custom user CLI commands in Flask

cli click command-group
Intermediate 7 steps
python 
import datetime
from dataclasses import dataclass
 
 

Parsing fixed-width records in Python

parsing generators dataclasses
Intermediate 8 steps
python 
from itertools import product
from dataclasses import dataclass
from decimal import Decimal
 

Generating product variants with itertools.product

cartesian-product dataclass decimal
Intermediate 7 steps
python 
from typing import Callable, Dict, Type
 
 
class PluginRegistry:

A decorator-based plugin registry in Python

decorators registry pattern factory
Intermediate 9 steps
python 
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
 

Building a signup endpoint in FastAPI

dependency-injection request-validation background-tasks
Intermediate 8 steps
python 
from django import forms
from django.utils import timezone
 
from .models import Reservation

Multi-field validation in a Django ModelForm

form validation cross-field validation modelform
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Merging sorted log files with heapq — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code