Package Exports

The package root re-exports the supported user-facing objects. Prefer these imports in application code.

from http_to_arrow import ArrowRecordContainer

http_to_arrow

Public exports for the http_to_arrow package.

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()

ArrowRecordContainer(schema=None, table=None, batch_size=128000, unknown_field_policy='drop', missing_field_policy='null', coercion_policy='coerce', case_insensitive_keys=True, eager_clear_accumulator=False, dictionary_encode=False, dictionary_cardinality_threshold=0.5, compact_on_materialize=False, batches=None) dataclass

Bases: BaseArrowRecordContainer, ArrowRecordContainerSettings

Batch incoming records into Arrow tables using an explicit or inferred schema.

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

def __init__(
    self,
    schema: pa.Schema | None = None,
    table: pa.Table | None = None,
    batch_size: int = 128000,  # Results in ~700MB batches in profiler test
    unknown_field_policy: UnknownFieldPolicy = "drop",
    missing_field_policy: MissingFieldPolicy = "null",
    coercion_policy: CoercionPolicy = "coerce",
    case_insensitive_keys: bool = True,
    eager_clear_accumulator: bool = False,
    dictionary_encode: bool = False,
    dictionary_cardinality_threshold: float = 0.5,
    compact_on_materialize: bool = False,
    batches: list[pa.RecordBatch] | None = None,
) -> None:
    self.schema = schema
    self.table = table
    self.batch_size = batch_size
    self.unknown_field_policy = unknown_field_policy
    self.missing_field_policy = missing_field_policy
    self.coercion_policy = coercion_policy
    self.case_insensitive_keys = case_insensitive_keys
    self.eager_clear_accumulator = eager_clear_accumulator
    self.dictionary_encode = dictionary_encode
    self.dictionary_cardinality_threshold = dictionary_cardinality_threshold
    self.compact_on_materialize = compact_on_materialize
    self.batches = deque() if batches is None else deque(batches)
    self.captured_extras = []
    self._schema_fields = ()
    self._schema_field_names = frozenset()
    self._uses_default_normalizer = False
    self._schema_explicit = False
    self._inferred_name_map = {}
    self._accumulator = {}
    self._current_count = 0
    self._pending_batch_rows = 0
    self._materialized_schema = None
    self._lock = threading.RLock()
    self.__post_init__()

batch_total_rows property

Total rows across pending batches plus the in-flight accumulator.

total_rows property

Total rows across the cached table, pending batches, and accumulator.

normalize_record(record)

Normalize records before schema matching and coercion.

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

def normalize_record(self, record: Mapping[str, Any]) -> dict[str, Any]:
    """Normalize records before schema matching and coercion."""
    return dict(record)

append(record)

Append a single record to the container.

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

def append(self, record: Mapping[str, Any]) -> None:
    """Append a single record to the container."""
    _appending.append(self, record)

extend(records)

Append multiple records to the container.

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

def extend(self, records: Iterable[Mapping[str, Any]]) -> None:
    """Append multiple records to the container."""
    _appending.extend(self, records)

flush()

Convert the current accumulator into a pending RecordBatch.

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

def flush(self) -> None:
    """Convert the current accumulator into a pending RecordBatch."""
    _materialization.flush(self)

to_table()

Materialize pending records and batches into a cached Arrow table.

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

def to_table(self) -> pa.Table:
    """Materialize pending records and batches into a cached Arrow table."""
    return _materialization.to_table(self)

incremental_flush(threshold=0)

Flush accumulated batches into the cached table when above threshold rows.

Unlike to_table() this is designed to be called periodically during streaming ingestion to bound memory. When the pending batch row count exceeds threshold, the batches are materialised into the cached table and the batch list is cleared.

Returns True when batches were actually flushed, False otherwise.

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

def incremental_flush(self, threshold: int = 0) -> bool:
    """Flush accumulated batches into the cached table when above *threshold* rows.

    Unlike ``to_table()`` this is designed to be called periodically during
    streaming ingestion to bound memory. When the pending batch row count
    exceeds *threshold*, the batches are materialised into the cached table
    and the batch list is cleared.

    Returns ``True`` when batches were actually flushed, ``False`` otherwise.
    """
    return _materialization.incremental_flush(self, threshold)

drain_batches()

Return completed pending batches and remove them from the container.

This releases ownership of every flushed RecordBatch without touching the in-flight accumulator or the cached table. After draining, :attr:batch_total_rows reflects only the rows still held in the accumulator.

Returns an empty list when no batches are pending.

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

def drain_batches(self) -> list[pa.RecordBatch]:
    """Return completed pending batches and remove them from the container.

    This releases ownership of every flushed ``RecordBatch`` without
    touching the in-flight accumulator or the cached table. After draining,
    :attr:`batch_total_rows` reflects only the rows still held in the
    accumulator.

    Returns an empty list when no batches are pending.
    """
    with self._lock:
        batches = self.batches
        self.batches = deque()
        self._pending_batch_rows = 0
    return list(batches)

flush_partial()

Flush the in-flight accumulator and return the resulting batch.

Unlike :meth:flush, this returns the newly created RecordBatch and removes it from the pending batch list so a later :meth:to_table call does not materialize it a second time. Returns None when the accumulator holds no rows.

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

def flush_partial(self) -> pa.RecordBatch | None:
    """Flush the in-flight accumulator and return the resulting batch.

    Unlike :meth:`flush`, this returns the newly created ``RecordBatch`` and
    removes it from the pending batch list so a later :meth:`to_table` call
    does not materialize it a second time. Returns ``None`` when the
    accumulator holds no rows.
    """
    with self._lock:
        if self._current_count == 0:
            return None
        self.flush()
        batch = self.batches.pop()
        self._pending_batch_rows -= batch.num_rows
    return batch

iter_batches()

Yield completed batches one at a time, releasing each as it is yielded.

Batches are yielded in FIFO order and removed from the container as they are produced, so memory is not retained for already-yielded batches. The in-flight accumulator is left untouched; call :meth:flush_partial first to emit a trailing short batch.

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

def iter_batches(self) -> Iterator[pa.RecordBatch]:
    """Yield completed batches one at a time, releasing each as it is yielded.

    Batches are yielded in FIFO order and removed from the container as they
    are produced, so memory is not retained for already-yielded batches. The
    in-flight accumulator is left untouched; call :meth:`flush_partial`
    first to emit a trailing short batch.
    """
    while True:
        with self._lock:
            if not self.batches:
                break
            batch = self.batches.popleft()
            self._pending_batch_rows -= batch.num_rows
        yield batch

to_polars_frame()

Materialize the container as a Polars DataFrame.

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

def to_polars_frame(self) -> pl.DataFrame:
    """Materialize the container as a Polars DataFrame."""
    return _materialization.to_polars_frame(self)

reset()

Clear accumulated data, batches, cached table, extras, and caches.

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

def reset(self) -> None:
    """Clear accumulated data, batches, cached table, extras, and caches."""
    _materialization.reset(self)

to_arrow()

Backward-compatible alias for materializing an Arrow table.

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

def to_arrow(self) -> pa.Table:
    """Backward-compatible alias for materializing an Arrow table."""
    return self.to_table()

to_polars()

Backward-compatible alias for materializing a Polars DataFrame.

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

def to_polars(self) -> pl.DataFrame:
    """Backward-compatible alias for materializing a Polars DataFrame."""
    return self.to_polars_frame()

clear()

Backward-compatible alias for resetting container state.

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

def clear(self) -> None:
    """Backward-compatible alias for resetting container state."""
    self.reset()

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