Read JSONL Records Line by Line

import json
import tempfile
from collections.abc import Iterator
from pathlib import Path
from typing import Any


class JsonlError(RuntimeError):
    def __init__(self, line_no: int, message: str) -> None:
        super().__init__(f"line {line_no}: {message}")
        self.line_no = line_no


def read_jsonl(
    path: Path,
    *,
    max_record_bytes: int = 64 * 1024,
) -> Iterator[dict[str, Any]]:
    if max_record_bytes < 2:
        raise ValueError("max_record_bytes must be at least 2")
    with path.open("r", encoding="utf-8") as handle:
        for line_no, raw in enumerate(handle, start=1):
            size = len(raw.encode("utf-8"))
            if size > max_record_bytes:
                raise JsonlError(
                    line_no,
                    f"record is {size} bytes, limit is {max_record_bytes}",
                )
            line = raw.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError as error:
                raise JsonlError(line_no, f"invalid JSON: {error.msg}") from error
            if not isinstance(record, dict):
                raise JsonlError(line_no, "record must be a JSON object")
            yield record


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "events.jsonl"
        path.write_text(
            '{"id": 1, "kind": "click"}\n'
            "\n"
            '{"id": 2, "kind": "view"}\n'
            '{"id": 3, "kind": }\n'
            '{"id": 4, "kind": "buy"}\n',
            encoding="utf-8",
        )
        good = 0
        try:
            for record in read_jsonl(path):
                good += 1
                print("record", record["id"])
        except JsonlError as error:
            print(f"stopped after {good} records: {error}")


if __name__ == "__main__":
    main()