API Reference

CrystalXMLSource¶

CrystalXMLSource(source: str | Path, *, row_tag: str = "Row",
                 engine: str = "auto", threads: int = 0,
                 memory: str | int | None = None,
                 field_mapping: dict[str, str] | None = None,
                 drop_fields: list[str] | None = None,
                 filter: dict | None = None,
                 field_types: dict[str, str] | None = None,
                 dictionary_columns: list[str] | None = None,
                 schema: list[str] | None = None,
                 auto_dict: bool = False,
                 use_mmap: bool = True,
                 batch_size: int = 1024)

Param	Type	Default	Description
`source`	`str \\| Path`	required	Path to CR XML file
`row_tag`	`str`	`"Row"`	XML tag for each record row

Returns: iterable of dict[str, str]

Raises: FileNotFoundError, XmlError (bad CR XML format)

RenameFields¶

RenameFields(mapping: dict[str, str])

Param	Type	Description
`mapping`	`dict[str, str]`	Old key → new key mapping

Fusable: yes | Picklable: yes

CastTypes¶

CastTypes(mapping: dict[str, type])

Param	Type	Description
`mapping`	`dict[str, type]`	Field name → target type (e.g. `int`, `float`, `bool`, `str`)

Fusable: yes | Picklable: yes

DropFields¶

DropFields(fields: list[str])

Param	Type	Description
`fields`	`list[str]`	Keys to remove from rows

Fusable: yes | Picklable: yes

FilterRows¶

FilterRows(predicate: Callable[[dict], bool] | Predicate])
FilterRows(field=, op=, value=)
FilterRows(field_a=, op=, field_b=)

Param	Type	Description
`predicate`	`Callable[[dict], bool]`	Return `True` to keep the row; also accepts an expression predicate (`col("x") > 1`, see `rypipe.expr`; `col` is re-exported as `crxml.col`)

Keyword filters support op="regex" (regular expression search against the string form of the field; invalid patterns raise at construction time). Expression predicates support matches(pattern) (regex) and between(lo, hi) (inclusive range) and compose with &, |, ~.

Fusable: yes | Picklable: no (unless a module-level function)

Pipeline¶

Pipeline(source: Iterable[dict], stages: list[Stage] | None = None,
         *, batch_size: int = 1000, workers: int | None = None)

Created implicitly via |. Not typically constructed directly. When constructed directly, stages must be a list.

Methods¶

Method	Signature	Description
`__or__`	`(self, stage) -> Pipeline`	Append a stage
`__iter__`	`(self) -> Iterator[dict]`	Iterate rows
`parallel`	`(self, workers=None, batch_size=1000)`	Return parallel variant

CrystalXMLSource.schema¶

source.schema() -> list[str]

Returns the field name keys from the first row. The source caches the first batch internally, so calling schema() before building a pipeline does not lose data.

Returns: Empty list [] if the source is empty.

from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml")
fields = src.schema()
print(fields)  # ['{Report.InvoiceNo}', '{Report.Amount}', ...]

to_pandas¶

to_pandas(pipeline: Pipeline, chunksize: int | None = None) -> pd.DataFrame

Param	Type	Default	Description
`pipeline`	`Pipeline`	,	Pipeline to consume
`chunksize`	`int \\| None`	`None`	Incremental chunk size

to_csv¶

to_csv(pipeline: Pipeline, path: str | Path, encoding: str = "utf-8",
       delimiter: str = ",", fieldnames: list[str] | None = None) -> None

Param	Type	Default	Description
`pipeline`	`Pipeline`	,	Pipeline to consume
`path`	`str \\| Path`	,	Output CSV path
`encoding`	`str`	`"utf-8"`	Output file encoding
`delimiter`	`str`	`","`	Field separator
`fieldnames`	`list[str] \\| None`	`None`	Explicit header; defaults to first record's keys

The header comes from the first record unless fieldnames is given. CR exports are ragged: fields that appear later but are missing from the header are omitted, and a UserWarning names them (once per field). Pass a fieldnames union to include them. Fields missing from a record are written as empty strings.

Exceptions¶

Typed exceptions from the Rust core, importable from crxml:

Exception	Raised when
`XmlError`	Input cannot be parsed (malformed XML, failed UTF-8 validation)
`PlanError`	Invalid pushdown plan kwargs (unknown op, unknown field type)
`MergeError`	Chunk merge conflict during multi-chunk/parallel/bounded parsing

from crxml import CrystalXMLSource, XmlError

try:
    CrystalXMLSource("export.xml").to_pandas()
except XmlError as e:
    print(f"bad input: {e}")

collect¶

collect(pipeline: Pipeline) -> list[dict]

Param	Type	Description
`pipeline`	`Pipeline`	Pipeline to materialize

discover_schema¶

discover_schema(source: str | Path, *, row_tag: str = "Details",
                field_mapping: dict[str, str] | None = None,
                drop_fields: list[str] | None = None,
                filter: dict | None = None,
                field_types: dict[str, str] | None = None,
                dictionary_columns: list[str] | None = None,
                schema: list[str] | None = None,
                auto_dict: bool = False) -> list[str]

Discover the frozen schema for a file. Scans once (full scan for ≤128 MB, else 16×2 MiB sampled windows in parallel) and returns column names in file order after applying field_mapping/drop_fields/filter etc.

Pass the result as CrystalXMLSource(..., schema=schema) to avoid per-file schema discovery and hit the explicit fast path.

Param	Type	Default	Description
`source`	`str \\| Path`	required	Path to CR XML file
`row_tag`	`str`	`"Details"`	XML tag for each record row

Returns: list[str] of field names in output order.