Streaming

The http_to_arrow.streaming module provides the async Arrow IPC streaming surface. See the Streaming IPC guide for a task-oriented walkthrough.

Tip

Import ArrowIPCStream and IPCStreamSink from http_to_arrow rather than from their implementation modules.

Async stream

streaming

Async Arrow IPC streaming for http_to_arrow.

ArrowIPCStream bridges an async row producer (for example an HTTP paginator) to an Arrow IPC byte stream. Rows are buffered into an :class:~http_to_arrow.main.ArrowRecordContainer, completed batches are serialized to Arrow IPC stream bytes via :class:~http_to_arrow._ipc.IPCStreamSink, and each serialized batch is released immediately so peak memory stays close to a single batch.

A bounded :class:asyncio.Queue decouples the producer from the consumer so a fast paginator never outruns batch serialization. Each :class:ArrowIPCStream instance is single-use: create a fresh stream per response.

ArrowIPCStream(schema, *, batch_size=128000, queue_maxsize=4, compression=None, dictionary_encode=False, dictionary_cardinality_threshold=0.5, eager_clear_accumulator=True)

Stream Arrow IPC bytes from an async row producer with backpressure.

Streaming requires an explicit schema because the Arrow IPC stream header is written before any rows are available. Inferred-schema mode and dictionary_encode=True are rejected because they can change batch types after the header schema is fixed.

Example

stream = ArrowIPCStream(schema=schema, compression="zstd")

async def produce() -> None: async for page in fetch_pages(): await stream.extend(page)

async for chunk in stream.ipc_chunks(producer=produce): ... # forward chunk to the HTTP response

Source code in src/http_to_arrow/src/http_to_arrow/streaming.py

def __init__(
    self,
    schema: pa.Schema,
    *,
    batch_size: int = 128_000,
    queue_maxsize: int = 4,
    compression: str | None = None,
    dictionary_encode: bool = False,
    dictionary_cardinality_threshold: float = 0.5,
    eager_clear_accumulator: bool = True,
) -> None:
    if schema is None:
        raise ValueError("ArrowIPCStream requires an explicit schema.")
    if dictionary_encode:
        raise ValueError(
            "ArrowIPCStream does not support dictionary_encode=True. The IPC "
            "stream header fixes the schema before batches are written, and "
            "dictionary promotion can change later batch types."
        )

    self._schema = schema
    self._compression = compression
    self._queue: asyncio.Queue[object] = asyncio.Queue(maxsize=queue_maxsize)
    self._ended = False
    self._container_kwargs: dict[str, Any] = {
        "schema": schema,
        "batch_size": batch_size,
        "dictionary_encode": dictionary_encode,
        "dictionary_cardinality_threshold": dictionary_cardinality_threshold,
        "eager_clear_accumulator": eager_clear_accumulator,
    }

put(row) async

Enqueue a single row, awaiting when the queue is full.

Source code in src/http_to_arrow/src/http_to_arrow/streaming.py

async def put(self, row: Mapping[str, Any]) -> None:
    """Enqueue a single row, awaiting when the queue is full."""
    await self._queue.put([row])

extend(rows) async

Enqueue a page of rows, awaiting when the queue is full.

Source code in src/http_to_arrow/src/http_to_arrow/streaming.py

async def extend(self, rows: Sequence[Mapping[str, Any]]) -> None:
    """Enqueue a page of rows, awaiting when the queue is full."""
    await self._queue.put(list(rows))

end() async

Signal that no more rows will be enqueued. Idempotent.

Source code in src/http_to_arrow/src/http_to_arrow/streaming.py

async def end(self) -> None:
    """Signal that no more rows will be enqueued. Idempotent."""
    if not self._ended:
        self._ended = True
        await self._queue.put(_SENTINEL)

ipc_chunks(producer=None) async

Yield Arrow IPC stream byte chunks until the producer signals end.

When producer is provided it is run as a concurrent task and this method signals :meth:end on its behalf, so the producer never needs to call it. Otherwise rows must be fed from a separate task via :meth:put/:meth:extend, and that task must call :meth:end when done.

If the producer raises, the partial stream is flushed and closed cleanly and the exception is re-raised after the trailing bytes are emitted.

Source code in src/http_to_arrow/src/http_to_arrow/streaming.py

async def ipc_chunks(
    self,
    producer: Callable[[], Awaitable[None]] | None = None,
) -> AsyncIterator[bytes]:
    """Yield Arrow IPC stream byte chunks until the producer signals end.

    When *producer* is provided it is run as a concurrent task and this
    method signals :meth:`end` on its behalf, so the producer never needs to
    call it. Otherwise rows must be fed from a separate task via
    :meth:`put`/:meth:`extend`, and that task must call :meth:`end` when done.

    If the producer raises, the partial stream is flushed and closed cleanly
    and the exception is re-raised after the trailing bytes are emitted.
    """
    container = ArrowRecordContainer(**self._container_kwargs)
    ipc_sink = IPCStreamSink(self._schema, compression=self._compression)
    producer_error: BaseException | None = None
    producer_task: asyncio.Task[None] | None = None

    async def _run_producer(run: Callable[[], Awaitable[None]]) -> None:
        nonlocal producer_error
        try:
            await run()
        except Exception as exc:  # surfaced to the consumer after cleanup
            producer_error = exc
        finally:
            await self.end()

    header = ipc_sink.header_bytes()
    if header:
        yield header

    if producer is not None:
        producer_task = asyncio.create_task(_run_producer(producer))

    try:
        while True:
            item = await self._queue.get()
            if item is _SENTINEL:
                break

            for row in cast("list[Mapping[str, Any]]", item):
                container.append(row)

            for batch in container.iter_batches():
                chunk = ipc_sink.write_batch(batch)
                del batch
                if chunk:
                    yield chunk

            # Cooperative yield. On the accumulation-only path (a page
            # smaller than the remaining batch_size completes no batch and
            # emits no chunk) a non-empty ``queue.get()`` never suspends, so
            # this is the only point that returns control to the event loop
            # and keeps the producer task from being starved. Benchmarked as
            # effectively free at realistic page sizes.
            await asyncio.sleep(0)

        final_batch = container.flush_partial()
        if final_batch is not None:
            chunk = ipc_sink.write_batch(final_batch)
            del final_batch
            if chunk:
                yield chunk

        closing = ipc_sink.close()
        if closing:
            yield closing

        if producer_error is not None:
            raise producer_error
    finally:
        if producer_task is not None and not producer_task.done():
            producer_task.cancel()
            try:
                await producer_task
            except asyncio.CancelledError:
                pass
            except Exception:
                pass

IPC stream sink

_ipc

Incremental Arrow IPC stream serialization for http_to_arrow.

IPCStreamSink wraps :func:pyarrow.ipc.new_stream around a :class:~http_to_arrow._drain.DrainableOutputStream so that each written RecordBatch can be serialized to Arrow IPC stream bytes and yielded immediately, keeping peak memory close to a single batch.

IPCStreamSink(schema, *, compression=None)

Serialize RecordBatch objects into Arrow IPC stream byte chunks.

The schema is written into the stream header on construction; call :meth:header_bytes to retrieve it, :meth:write_batch for each batch, and :meth:close to emit the end-of-stream marker.

Example

sink = IPCStreamSink(schema, compression="zstd") chunks = [sink.header_bytes()] chunks.append(sink.write_batch(batch)) chunks.append(sink.close()) table = pa.ipc.open_stream(pa.BufferReader(b"".join(chunks))).read_all()

Source code in src/http_to_arrow/src/http_to_arrow/_ipc.py

def __init__(self, schema: pa.Schema, *, compression: str | None = None) -> None:
    self._sink = DrainableOutputStream()
    options = (
        pa.ipc.IpcWriteOptions(compression=cast("Any", compression))
        if compression is not None
        else None
    )
    # ``DrainableOutputStream`` is a duck-typed write-only sink; PyArrow wraps
    # it through its Python file interface even though the stubs only admit
    # native files.
    self._writer = pa.ipc.new_stream(
        cast("Any", self._sink), schema, options=options
    )
    self._header = self._sink.drain()
    self._closed = False

closed property

Whether the stream writer has been closed.

header_bytes()

Return the Arrow IPC stream header bytes captured at construction.

PyArrow may buffer the schema message until the first write_batch or close, so this can be empty. The bytes are always emitted in stream order, so concatenating header_bytes() with every write_batch result and close produces a valid Arrow IPC stream.

Source code in src/http_to_arrow/src/http_to_arrow/_ipc.py

def header_bytes(self) -> bytes:
    """Return the Arrow IPC stream header bytes captured at construction.

    PyArrow may buffer the schema message until the first ``write_batch`` or
    ``close``, so this can be empty. The bytes are always emitted in stream
    order, so concatenating ``header_bytes()`` with every ``write_batch``
    result and ``close`` produces a valid Arrow IPC stream.
    """
    return self._header

write_batch(batch)

Serialize batch and return its Arrow IPC stream bytes.

Source code in src/http_to_arrow/src/http_to_arrow/_ipc.py

def write_batch(self, batch: pa.RecordBatch) -> bytes:
    """Serialize *batch* and return its Arrow IPC stream bytes."""
    if self._closed:
        raise RuntimeError("IPCStreamSink is closed")
    self._writer.write_batch(batch)
    return self._sink.drain()

close()

Close the writer and return the trailing end-of-stream bytes.

Source code in src/http_to_arrow/src/http_to_arrow/_ipc.py

def close(self) -> bytes:
    """Close the writer and return the trailing end-of-stream bytes."""
    if not self._closed:
        self._writer.close()
        self._closed = True
    return self._sink.drain()