jsonl.py — anthropic-sdk-python Source File

Architecture documentation for jsonl.py, a python file in the anthropic-sdk-python codebase. 4 imports, 0 dependents.

File python StreamProtocol EventLogic 4 imports 2 classes

Entity Profile

StreamProtocol→ EventLogic→ jsonl.py — anthropic-sdk-python Source File

Dependency Diagram

graph LR
  e75927c6_4f5f_0cbb_8978_9029ad9f28aa["jsonl.py"]
  508ad9e1_b10f_2976_4503_76049ea413ab["_models"]
  e75927c6_4f5f_0cbb_8978_9029ad9f28aa --> 508ad9e1_b10f_2976_4503_76049ea413ab
  28b0c811_20f6_fc4a_4b48_7fb9e87bf7e5["json"]
  e75927c6_4f5f_0cbb_8978_9029ad9f28aa --> 28b0c811_20f6_fc4a_4b48_7fb9e87bf7e5
  37c05070_ca59_d596_7250_de9d1939227f["typing_extensions"]
  e75927c6_4f5f_0cbb_8978_9029ad9f28aa --> 37c05070_ca59_d596_7250_de9d1939227f
  9c26e8a9_1ad2_1174_876a_1fc500ce0eaf["httpx"]
  e75927c6_4f5f_0cbb_8978_9029ad9f28aa --> 9c26e8a9_1ad2_1174_876a_1fc500ce0eaf
  style e75927c6_4f5f_0cbb_8978_9029ad9f28aa fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

from __future__ import annotations

import json
from typing_extensions import Generic, TypeVar, Iterator, AsyncIterator

import httpx

from .._models import construct_type_unchecked

_T = TypeVar("_T")


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,
            )
// ... (64 more lines)