ndjson

values = byte_stream.pipe_through_channel(ndjson.decode())

encode_string() -> Channel[tuple[str, ...], E | NdjsonError, D, tuple[object, ...], E, D]

def encode_string[E = Never, D = None]() -> Channel[
    tuple[str, ...], E | NdjsonError, D, tuple[object, ...], E, D
]:
    """Write each chunk of values as one string of lines, each ending with
    ``\\n``. A value with no JSON form (a set, an object, ``NaN``) fails
    with a ``"Pack"`` error."""

    def encoded(chunk: tuple[object, ...]) -> AnyPull:
        try:
            text = "\n".join(map(stringify, chunk)) + "\n"
        except STRINGIFY_ERRORS as cause:
            return _FailCause(fail_cause(NdjsonError("Pack", cause)))
        return _Succeed((text,))

    channel: Channel[tuple[str, ...], E | NdjsonError, D, tuple[object, ...], E, D] = (
        from_transform(lambda upstream, _scope: _Succeed(upstream.flat_map(encoded)))
    )
    return channel

encode() -> Channel[tuple[bytes, ...], E | NdjsonError, D, tuple[object, ...], E, D]

def encode[E = Never, D = None]() -> Channel[
    tuple[bytes, ...], E | NdjsonError, D, tuple[object, ...], E, D
]:
    """``encode_string`` as UTF-8 bytes."""
    strings: Channel[tuple[str, ...], E | NdjsonError, D, tuple[object, ...], E, D] = (
        encode_string()
    )
    return strings.map(lambda chunk: tuple(text.encode() for text in chunk))

decode_string(*, ignore_empty_lines=False) -> Channel[tuple[Json, ...], E | NdjsonError, D, tuple[str, ...], E, D]

def decode_string[E = Never, D = None](
    *, ignore_empty_lines: bool = False
) -> Channel[tuple[Json, ...], E | NdjsonError, D, tuple[str, ...], E, D]:
    """Parse chunks of text as NDJSON. The first line that is not JSON fails
    the channel (``"InvalidJson"``) after the values before it are written;
    blank lines fail too unless ``ignore_empty_lines``."""

    def transform(upstream: AnyPull, scope: CloseableScope) -> Any:
        # ``split_lines`` writes a line after its input has ended only when
        # that line has no line break.
        ended = False

        def done(leftover: object) -> AnyPull:
            nonlocal ended
            ended = True
            return end_with(leftover)

        def failed(cause: Cause[Any]) -> AnyPull:
            return _FailCause(cause)

        source = match_effect(upstream, on_success=_Succeed, on_failure=failed, on_done=done)

        def step(lines: tuple[str, ...], out: list[Json]) -> NdjsonError | None:
            for line in lines:
                error = _parse_line(line, out, ignore_empty_lines, ended)
                if error is not None:
                    return error
            return None

        return (
            split_lines()
            ._transform(source, scope)
            .map(lambda lines: _decode_pull(lines, step, _flush_nothing))
        )

    channel: Channel[tuple[Json, ...], E | NdjsonError, D, tuple[str, ...], E, D] = from_transform(
        transform
    )
    return channel

decode(*, ignore_empty_lines=False, max_line_bytes=None) -> Channel[tuple[Json, ...], E | NdjsonError, D, tuple[bytes, ...], E, D]

def decode[E = Never, D = None](
    *, ignore_empty_lines: bool = False, max_line_bytes: int | None = None
) -> Channel[tuple[Json, ...], E | NdjsonError, D, tuple[bytes, ...], E, D]:
    """Parse chunks of UTF-8 bytes as NDJSON; a leading byte-order mark is
    skipped. The first bad line fails the channel after the values before
    it are written: ``"InvalidUtf8"``, ``"InvalidJson"``, or
    ``"LineTooLarge"`` as soon as a line passes ``max_line_bytes`` (its
    bytes without the line break; ``None``, the default, is unbounded),
    without reading the rest of it."""
    limit = math.inf if max_line_bytes is None else max_line_bytes

    def build(upstream: AnyPull) -> AnyPull:
        lines = _ByteLines(ignore_empty_lines, limit)
        return _decode_pull(upstream, lines.step, lines.flush)

    channel: Channel[tuple[Json, ...], E | NdjsonError, D, tuple[bytes, ...], E, D] = (
        from_transform(lambda upstream, _scope: _Sync(lambda: build(upstream)))
    )
    return channel

duplex(channel, *, ignore_empty_lines=False, max_line_bytes=None) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def duplex(
    channel: Channel[Any, Any, Any, Any, Any, Any, Any],
    *,
    ignore_empty_lines: bool = False,
    max_line_bytes: int | None = None,
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Wrap a channel of bytes both ways: the values written to it go out as
    NDJSON bytes, and the bytes it writes come back parsed (``decode``)."""
    require_channel(channel, "channel")
    outgoing: Channel[Any, Any, Any, Any, Any, Any, Any] = encode()
    incoming: Channel[Any, Any, Any, Any, Any, Any, Any] = decode(
        ignore_empty_lines=ignore_empty_lines, max_line_bytes=max_line_bytes
    )
    return outgoing.pipe_to(channel).pipe_to(incoming)

duplex_string(channel, *, ignore_empty_lines=False) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def duplex_string(
    channel: Channel[Any, Any, Any, Any, Any, Any, Any],
    *,
    ignore_empty_lines: bool = False,
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """``duplex`` for a channel of text."""
    require_channel(channel, "channel")
    outgoing: Channel[Any, Any, Any, Any, Any, Any, Any] = encode_string()
    incoming: Channel[Any, Any, Any, Any, Any, Any, Any] = decode_string(
        ignore_empty_lines=ignore_empty_lines
    )
    return outgoing.pipe_to(channel).pipe_to(incoming)