Stream JSON Lines records with a generator
Python
import json
from collections.abc import Iterator
from pathlib import Path
def iter_jsonl(path: str | Path) -> Iterator[dict]:
"""Yield one decoded object per non-empty line, with error positions."""
with open(path, encoding="utf-8") as handle:
for number, line in enumerate(handle, start=1):
text = line.strip()
if not text:
continue
try:
record = json.loads(text)
except json.JSONDecodeError as exc:
raise ValueError(f"{path}:{number}: invalid JSON: {exc}") from exc
if not isinstance(record, dict):
raise ValueError(f"{path}:{number}: expected a JSON object")
yield record
def write_jsonl(path: str | Path, records: list[dict]) -> None:
"""Append records in JSON Lines format, one object per line."""
with open(path, "a", encoding="utf-8") as handle:
for record in records:
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
if __name__ == "__main__":
write_jsonl("events.jsonl", [{"id": 1}, {"id": 2}])
print(list(iter_jsonl("events.jsonl")))