Container

The http_to_arrow.main module contains the primary ArrowRecordContainer implementation.

Tip

Most users should import ArrowRecordContainer from http_to_arrow rather than from http_to_arrow.main.

main

Shared Arrow-backed record containers for ETL-style ingestion flows.

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