JSONLDecoder Class — anthropic-sdk-python Architecture

Source Code

src/anthropic/_decoders/jsonl.py lines 13–67

class JSONLDecoder(Generic[_T]):
    """A decoder for [JSON Lines](https://jsonlines.org) format.

    This class provides an iterator over a byte-iterator that parses each JSON Line
    into a given type.
    """

    http_response: httpx.Response
    """The HTTP response this decoder was constructed from"""

    def __init__(
        self,
        *,
        raw_iterator: Iterator[bytes],
        line_type: type[_T],
        http_response: httpx.Response,
    ) -> None:
        super().__init__()
        self.http_response = http_response
        self._raw_iterator = raw_iterator
        self._line_type = line_type
        self._iterator = self.__decode__()

    def close(self) -> None:
        """Close the response body stream.

        This is called automatically if you consume the entire stream.
        """
        self.http_response.close()

    def __decode__(self) -> Iterator[_T]:
        buf = b""
        for chunk in self._raw_iterator:
            for line in chunk.splitlines(keepends=True):
                buf += line
                if buf.endswith((b"\r", b"\n", b"\r\n")):
                    yield construct_type_unchecked(
                        value=json.loads(buf),
                        type_=self._line_type,
                    )
                    buf = b""

        # flush
        if buf:
            yield construct_type_unchecked(
                value=json.loads(buf),
                type_=self._line_type,
            )

    def __next__(self) -> _T:
        return self._iterator.__next__()

    def __iter__(self) -> Iterator[_T]:
        for item in self._iterator:
            yield item