ndjson.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/ollama/ndjson.py
 
"""
NDJSON (Newline-Delimited JSON) stream parser.
 
Purpose
-------
Reusable utility for parsing chunked NDJSON HTTP responses into a stream
of dicts. Used by :class:`providers.ollama.client.OllamaClient` and
reusable by future providers (LM Studio, vLLM).
 
Responsibilities
----------------
- :func:`parse_ndjson_stream` — consume an arbitrary-chunk-size byte
  iterator, accumulate a line buffer, and yield one dict per complete
  JSON line. Handles: chunks split across JSON boundaries, blank lines,
  malformed JSON lines, and final un-terminated lines.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — malformed JSON line logged and skipped
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Input: any ``Iterator[bytes]`` — HTTP response body iterated in
  arbitrary chunk sizes.
- Output: ``Iterator[dict]`` — one dict per valid JSON line.
- A ``json.JSONDecodeError`` on any single line is logged at L2 then
  skipped; the remaining stream continues normally.
- An empty iterator yields nothing.
- A final chunk without a trailing ``\\n`` is flushed after the iterator
  is exhausted.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import json
from collections.abc import Iterator
 
from oc_diagnostics import _dbg
 
 
def parse_ndjson_stream(byte_iter: Iterator[bytes]) -> Iterator[dict]:
    """Parse a chunked NDJSON HTTP response body into a stream of dicts.
 
    Each element of *byte_iter* may contain zero, one, or multiple
    ``\\n``-terminated JSON lines, or a partial line that continues in
    the next element.
 
    Args:
        byte_iter: Raw HTTP body chunk iterator (e.g. an iterated
            ``http.client.HTTPResponse``).
 
    Yields:
        Parsed dicts, one per complete, valid JSON line.
    """
    func = "parse_ndjson_stream"
    buf = bytearray()
 
    for chunk in byte_iter:
        buf.extend(chunk)
        while b"\n" in buf:
            line_bytes, buf = buf.split(b"\n", 1)
            line = line_bytes.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as exc:
                _dbg("OC_CHAT.CLIENT", func,
                     f"malformed JSON line ({exc}): {line[:80]!r}", 2)
 
    # Flush any remaining data (final un-terminated line)
    line = buf.strip()
    if line:
        try:
            yield json.loads(line)
        except json.JSONDecodeError as exc:
            _dbg("OC_CHAT.CLIENT", func,
                 f"malformed final line ({exc}): {line[:80]!r}", 2)
 
 
__all__ = ["parse_ndjson_stream"]