Python API¶

rypipe-python builds a mixed Rust/Python package. The public API lives in the rypipe package; _rypipe is the low-level Rust extension that adapter packages build on.

rypipe itself does not ship any format parsers. Install a separate adapter package and import it; the adapter registers itself with rypipe so the high-level read API works. Adapters can also expose a Source subclass and get the pipeline/stage/sink API for free.

Building the Python module¶

export PYO3_PYTHON=/path/to/python3.12
maturin develop --release

maturin builds crates/rypipe-python/Cargo.toml and installs both the rypipe Python package and the _rypipe Rust extension.

Public API (import rypipe)¶

rypipe.Source¶

Abstract base class for row-oriented file sources. Adapter packages subclass it and implement _read_arrow. Once they do, users get pipelines, stages, and sinks with no extra work.

from rypipe import Source

class MySource(Source):
    def _read_arrow(self, plan_overrides=None):
        # Build plan from construction kwargs + overrides, call the parser,
        # return a pyarrow.Table.
        ...

rypipe.Adapter¶

Even simpler: subclass Adapter and implement only read(path, **kwargs). Plan kwargs are merged and passed through automatically.

from rypipe import Adapter

class CsvAdapter(Adapter):
    def read(self, path, **kwargs):
        return _rypipe_csv.read_csv(path, **kwargs)

source = CsvAdapter("data.csv")

A Source exposes:

• Row iteration: for row in source
• Table export: source.to_arrow(), source.to_pandas(), source.to_polars(), source.to_parquet(path)
• Pipeline operator: source | RenameFields(...)
• Caching: source.clear_cache()

Pipeline stages¶

rypipe ships the same fusable stages crxml used. Stages that rename, drop, cast, or filter constants are pushed into the Rust parse loop when the source supports plan kwargs.

from rypipe import RenameFields, DropFields, CastTypes, FilterRows

pipeline = (
    source
    | RenameFields({"old_name": "new_name"})
    | DropFields(["internal_id"])
    | CastTypes({"amount": float, "qty": int})
    | FilterRows(field="status", op="==", value="active")
)

CastTypes accepts Python callables (int, float, bool, str). When the callable maps to a Rust type (int64, float64, bool), it is fused into the Rust parse loop.

FilterRows supports both constant filters and column-to-column comparisons:

FilterRows(field="status", op="==", value="active")
FilterRows(field_a="amount", op=">", field_b="threshold")

Supported ops: ==, !=, >, <, >=, <=.

Pipeline sinks¶

from rypipe import collect, to_arrow, to_dataframe, to_csv, to_parquet

rows = collect(pipeline)
table = to_arrow(pipeline)
df = to_dataframe(pipeline)
to_csv(pipeline, "out.csv")
to_parquet(pipeline, "out.parquet")

Sinks try the fused Arrow path first and fall back to dict iteration when the pipeline ends with a generic stage.

rypipe.read¶

Single entry point for all registered adapters.

import rypipe
import my_adapter  # registers the "myfmt" adapter

table = rypipe.read("data.myfmt")

# Same call with all common options:
table = rypipe.read(
    "data.myfmt",
    format="myfmt",          # inferred from extension when omitted
    fields={"amount": "float64", "qty": "int64"},
    dictionary=["status"],
    filter={"field": "status", "op": "==", "value": "active"},
    schema=["id", "status", "amount"],
    auto_dict=False,
    use_mmap=False,
    prefault=False,
)

Returns a pyarrow.Table.

You can also pass an adapter object directly:

table = rypipe.read("data.myfmt", adapter=my_adapter, row_tag="Row")

rypipe.read_par¶

Convenience wrapper that passes chunks to the adapter.

table = rypipe.read_par("data.myfmt", chunks=8, fields={"amount": "float64"})

rypipe.read_stream¶

Convenience wrapper that passes a memory budget to the adapter. memory accepts an int (bytes) or a human-readable string such as "128MiB".

table = rypipe.read_stream("huge.myfmt", memory="500MiB", row_tag="Row")

Format auto-detection¶

rypipe.read infers the adapter from the file extension when format is not provided, but only for extensions registered by an installed adapter package. If no adapter is registered, pass format= explicitly or install the adapter.

Exceptions¶

Low-level API (import _rypipe)¶

_rypipe is the Rust extension that adapter packages build on. It exposes the shared exceptions and Rust helper functions; adapter packages implement the actual read functions and call these helpers from their own PyO3 code.

Exceptions¶

• _rypipe.ParseError
• _rypipe.XmlError
• _rypipe.PlanError
• _rypipe.MergeError

Adapter code raises these so users can catch them through rypipe as well.

Rust helpers (used from adapter crates)¶

Adapter crates written in Rust use rypipe_python directly:

use rypipe_python::{execution_plan_from_kwargs, record_batches_to_pyarrow_table};

execution_plan_from_kwargs converts Python kwargs into a rypipe_core::ExecutionPlan. record_batches_to_pyarrow_table turns a slice of Arrow RecordBatches into a single pyarrow.Table.

Plan kwargs¶

All public read functions and Source constructors accept the same pushdown kwargs, which are passed through to the adapter.

Filters¶

Constant equality/inequality (evaluated per-row during parse):

filter={"field": "status", "op": "==", "value": "active"}
filter={"field": "status", "op": "!=", "value": "archived"}

Column-to-column comparison (evaluated after the table is assembled):

filter={"field_a": "amount", "op": ">", "field_b": "threshold"}

Supported comparison ops: >, <, >=, <=, ==, !=.

See also¶

• Rust API: the Rust engine and Pipeline API.
• Writing a format adapter: adding formats as separate packages.
• Architecture: design overview.