Python API Reference¶

rypipe.read(
    path,                     # str | PathLike: file path
    *,
    format=None,              # str | None: adapter name
    adapter=None,             # object with read() method
    **kwargs,                 # forwarded to the adapter
) -> pyarrow.Table

import rypipe
import crxml  # registers the crxml adapter

# Format inferred from extension
table = rypipe.read("report.xml", row_tag="Details")

# Format specified explicitly
table = rypipe.read("data.txt", format="crxml")

# Adapter passed directly
from crxml import CrystalXMLAdapter
table = rypipe.read("report.xml", adapter=CrystalXMLAdapter(), row_tag="Details")

rypipe.read_par(
    path,
    *,
    chunks=4,                 # int: number of parallel chunks
    **kwargs,
) -> pyarrow.Table

rypipe.read_stream(
    path,
    *,
    memory="64MiB",           # int | str: memory budget
    **kwargs,
) -> pyarrow.Table

rypipe.read_batches(
    path,
    *,
    memory="64MiB",           # int | str: memory budget
    batch_size=None,          # int | None: rows per batch
    **kwargs,
) -> Iterator[pyarrow.RecordBatch]

rypipe.iter_record_batches(
    path,
    *,
    format=None,              # str | None: adapter name
    adapter=None,             # object with read() method
    memory="64MiB",           # int | str: memory budget
    batch_size=None,          # int | None: rows per batch
    **kwargs,
) -> Iterator[pyarrow.RecordBatch]

rypipe.register_adapter(
    name,                     # str: adapter name
    adapter,                  # object with read() method
    extensions=None,          # Iterable[str] | None: file extensions
) -> None

rypipe.resolve_engine(
    file_size,                # int: file size in bytes (required)
    *,                        # keyword-only from here
    memory=None,              # int | str | None: memory budget (e.g. "64MiB")
    threads=None,             # int | None: number of threads (must be >= 1 if provided)
    schema=None,              # list[str] | None: projected column names, in order
    has_parallel=True,        # bool: adapter has parallel support
    has_columnar=True,        # bool: adapter has columnar support
) -> str

class Source(ABC):
    def __init__(
        self,
        path,                          # str | Path
        *,
        field_mapping=None,            # dict[str, str]
        drop_fields=None,              # list[str]
        filter=None,                   # dict | None
        field_types=None,              # dict[str, str]
        dictionary_columns=None,       # list[str]
        schema=None,                   # list[str]: project exactly these columns, in this order
        auto_dict=False,               # bool
        strict_types=False,            # bool: reject malformed data instead of nulling
        observer=None,                 # dict[str, callable] | None: row observer hooks
        use_mmap=True,                 # bool
        batch_size=1024,               # int: Python-side batch size
        **reader_options,              # adapter-specific reader options
    )

@abstractmethod
def _read_arrow(self, plan_overrides: dict | None = None) -> pyarrow.Table:
    ...

# Standard (materializes full table)
source.to_pandas(dtype_backend="pyarrow")  # Arrow-backed dtypes (default)
source.to_pandas(dtype_backend="numpy")    # NumPy-backed dtypes

# Stream parsing; the returned DataFrame still holds all output
source.to_pandas(memory="64MiB")
source.to_pandas(memory="64MiB", threads=16)  # parallel streaming

source.to_parquet("output.parquet")
source.to_parquet("output.parquet", compression="snappy")
source.to_parquet("output.parquet", compression="zstd", compression_level=9)
source.to_parquet("output.parquet", row_group_size=100_000)

# Stream when the adapter provides a streaming reader
source.to_parquet("output.parquet", memory="64MiB")
source.to_parquet("output.parquet", memory="64MiB", threads=16)  # parallel

crxml.discover_schema(
    source,                     # str | Path: file path
    *,
    row_tag="Details",          # str: row element name
    field_mapping=None,         # dict[str, str] | None
    drop_fields=None,           # list[str] | None
    filter=None,                # dict | None
    field_types=None,           # dict[str, str] | None
    dictionary_columns=None,    # list[str] | None
    schema=None,                # list[str] | None
    auto_dict=False,            # bool
) -> list[str]

class CrystalXMLSource(Source):
    def __init__(
        self,
        source,                          # str | Path
        *,
        row_tag="Row",                   # str: XML element name for one row
        engine="auto",                   # "auto" | "stream" | "columnar" | "parallel"
        threads=None,                    # int | None: adapter default
        memory=None,                     # str | int | None: memory bound
        chunks=None,                     # int | None: number of parallel chunks
        max_split_chunks=None,           # int | None: max split chunks
        min_chunk_bytes=None,            # int | None: split chunk-size floor
        # ... plus all Source.__init__ kwargs ...
    )

class Adapter(Source):
    def read(self, path: str, **kwargs) -> pyarrow.Table:
        raise NotImplementedError

class Pipeline:
    def __init__(self, source, stages=None, *, batch_size=1024):
        """Create a pipeline. Usually built via source | stage, not directly."""

    def __or__(self, stage) -> Pipeline:
        """Append a stage and return a new Pipeline."""

    def __iter__(self) -> Iterator[dict]:
        """Iterate rows as dicts."""

    def to_arrow(self) -> pyarrow.Table:
        """Materialize and cache the transformed result."""

    def schema(self) -> list[str]:
        """Return output column names from the materialized Arrow result."""

    def clear_cache(self) -> None:
        """Drop the cached transformed result."""

    def iter_arrow_batches(self, batch_size=None) -> Iterator[RecordBatch]:
        """Yield Arrow RecordBatch objects."""

    def iter_record_batches(self, memory="64MiB", batch_size=None) -> Iterator[RecordBatch]:
        """Stream batches with bounded parser memory when supported."""

    def to_pandas(self, memory=None, dtype_backend="pyarrow", **kwargs) -> pd.DataFrame:
        """Convert to pandas. Pass memory= for streaming."""

    def to_polars(self, memory=None, **kwargs) -> pl.DataFrame:
        """Convert to Polars. Pass memory= for streaming."""

    def to_parquet(self, path, memory=None, **kwargs) -> None:
        """Write to Parquet. Pass memory= for streaming."""

RenameFields(mapping: dict[str, str])

DropFields(fields: list[str])

CastTypes(mapping: dict[str, Callable])

FilterRows(
    predicate=None,            # Callable or rypipe.expr.Predicate
    *,
    field=None,                # str: column name (constant filter)
    op=None,                   # str: operator
    value=None,                # str: value (constant filter)
    field_a=None,              # str: left column (comparison)
    field_b=None,              # str: right column (comparison)
    is_null=None,              # bool: True keeps null/missing rows, False drops them
    is_type=None,              # str: keep rows where field matches this type
)

import crxml

crxml.FilterRows(crxml.col("amount") > 100)
crxml.FilterRows((crxml.col("age") >= 18) & crxml.col("name").startswith("A"))

FilterRowsAny(*filters: FilterRows)  # requires >= 2 filters

FilterRowsAll(*filters: FilterRows)  # requires >= 2 filters

from rypipe.stages import ObservedStage

class CountRejected(ObservedStage):
    def __init__(self):
        self.count = 0

    def observer_hooks(self):
        return {"on_row_rejected": self._count}

    def _count(self, row_index):
        self.count += 1

FilterRowsNot(inner: FilterRows)  # exactly 1 filter

rypipe.to_csv(
    pipeline,                  # Iterable[dict]
    path,                      # str | Path
    encoding="utf-8",          # str
    delimiter=",",             # str
    fieldnames=None,           # list[str] | None
) -> None