中文
devexamples

Parse Newline-Delimited JSON Streams

Decode newline-delimited JSON records from a file or socket one line at a time, reporting the offending line when a record fails to parse.

PythonData Integrity

Code

snippet
import io
import json
from typing import Any, Iterator


def iter_json_lines(stream: io.TextIOBase) -> Iterator[Any]:
    for line_number, raw in enumerate(stream, start=1):
        text = raw.strip()
        if not text:
            continue
        try:
            yield json.loads(text)
        except json.JSONDecodeError as error:
            raise ValueError(
                f"invalid JSON on line {line_number}: {error.msg}"
            ) from error


payload = io.StringIO('{"id": 1}\n{"id": 2}\n')
records = list(iter_json_lines(payload))
print(len(records))

Related Examples