corral API reference ¶

Build a :class:Package from a source path/URL.

Dispatch order (per docs/architecture.md §6.1 + the :mod:corral.io package docstring):

1. Resolve ``engine`` (default: registry default).
2. Resolve ``spec`` — explicit :class:`DataPackage` wins;
   else look for a ``datapackage.json`` alongside ``source``;
   else synthesise a minimal spec from whatever resources
   are discovered.
3. Resolve ``source`` to a :class:`ResourceListing`:

    a. explicit ``format=`` short-circuits sniffing;
    b. URL scheme (``duckdb://``, ``s3://``, ...);
    c. file extension (``.csv``, ``.parquet``, ``.duckdb``,
       ``.csv.zip``);
    d. directory-of-known-formats walk (``csv/``,
       ``parquet/``) — directories have no extension to
       dispatch on, so we walk children and ask each
       matching adapter to scan one. Mirrors
       :func:`corral.validation.structural._scan_directory_of_known_formats`
       so the structural validator and the package loader
       can't drift apart on "what counts as a directory of
       tables".
    e. ``probe`` chain (any adapter that recognises the
       source);
    f. :class:`~corral.io.FormatNotDetected`.

4. For each :class:`~corral.io.base.ResourceRef`, ask
   the owning adapter to ``read`` it lazily through
   ``engine``, then wrap the resulting
   :class:`~corral.engines.base.TableExpr` in a
   :class:`~corral.dataset.Table`. The spec is
   consulted for the matching
   :class:`~corral.spec.model.Schema` so the validation
   layer doesn't have to re-load it.

Parameters:

Returns:

Raises:

Examples:

>>> from corral.fixtures import sample
>>> from corral.dataset import Package
>>> from corral.engines.ibis_engine import IbisEngine
>>> pkg = Package.from_source(
...     sample.csv_dir(),
...     engine=IbisEngine(),
...     spec=sample.DATAPACKAGE,
...     tables=["book"],
... )
>>> list(pkg.keys())
['book']

Build a :class:Package from already-constructed :class:Table instances.

Useful for tests, ad-hoc compositions, and adapter-less in-memory sources. When spec is not provided, a minimal one is synthesised so :meth:validate and the dict-like surface still work.

Parameters:

Returns:

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Package, Table
>>> e = IbisEngine()
>>> link = Table(name="link", expr=e.from_records([{"id": 1}]), engine=e)
>>> p = Package.from_tables({"link": link})
>>> "link" in p
True

Run the selected validation passes and return the combined report.

Order: structural → schema (per table) → foreign_keys → sync_state. Each pass appends issues to the same report. Skip any pass by passing False for its flag.

Sync-state contract: after a clean FK pass, the :class:DirtyTracker (when attached) stamps each non-violating FK via :meth:DirtyTracker.stamp_fk_from_exprs so subsequent edits surface sync.fk_stale instead of silently writing the broken FK. The sync_state pass then consults :meth:DirtyTracker.check to report any stamps that no longer match the live tables. No-op when dirty_tracker is None.

Parameters:

Returns:

Examples:

>>> from corral.fixtures import sample
>>> from corral.dataset import Package
>>> from corral.engines.ibis_engine import IbisEngine
>>> pkg = Package.from_source(
...     sample.csv_dir(),
...     engine=IbisEngine(),
...     spec=sample.DATAPACKAGE,
...     tables=["book", "author"],
... )
>>> r = pkg.validate()
>>> isinstance(r.issues, list)
True

Persist every table in the package to dest.

Sync-state contract: before writing, consults two signals per table — the per-:class:Table dirty flag (always available) and :meth:DirtyTracker.is_table_dirty (when a tracker is attached). EITHER signal flags the table as stale — the union is the conservative thing to surface. Stale tables trigger :class:OutOfSyncWarning, or :class:OutOfSyncError under strict_sync=True. When no tracker is attached, the per-table flag alone is consulted, so a caller can still flag a mutation via :meth:Table.invalidate without installing the tracker.

Format dispatch goes through :mod:corral.io exactly as on the read path:

• format explicit → adapter resolved by name;
• else inferred from dest extension;
• else default to "parquet" when dest looks like a directory.

For partitioned/multi-table formats (parquet directory, duckdb file), each :class:Table writes to a per-table location under dest. The exact layout matches the read-side scan so a round-trip via :meth:from_source returns the same logical package. A zip (.zip / .csv.zip, or format="zip") holds one flat <table>.csv member per table plus a datapackage.json with inlined schemas, and is written atomically (staged beside dest, then renamed over it).

Parameters:

Raises:

Examples:

Roundtrip the bundled sample fixture through parquet::

>>> import tempfile, pathlib
>>> from corral.fixtures import sample
>>> from corral.dataset import Package
>>> from corral.engines.ibis_engine import IbisEngine
>>> with tempfile.TemporaryDirectory() as tmp:
...     pkg = Package.from_source(
...         sample.csv_dir(),
...         engine=IbisEngine(),
...         spec=sample.DATAPACKAGE,
...         tables=["book"],
...     )
...     out = pathlib.Path(tmp) / "out.pkg"
...     pkg.write(out, format="parquet")
...     out.exists()
True

Return self[name].count() or None if absent / transiently uncountable.

Convenience used by previews (notebook _repr_html_, CLI info, HTTP /networks/{id}) where a missing or transiently-broken table should degrade to a friendly “?” rather than crash the surrounding render.

The contract: returns None when the table is absent OR when table.count() raises any :class:Exception (e.g. a transient backend hiccup on an ibis-against-remote-duckdb connection). Hard exceptions in surrounding code still propagate — this only shields the count call itself.

Examples:

>>> from corral.dataset import Package, Table
>>> from corral.engines.ibis_engine import IbisEngine
>>> e = IbisEngine()
>>> pkg = Package.from_tables({"x": Table(name="x", expr=e.from_records([{"a": 1}]), engine=e)})
>>> pkg.safe_count("x")
1
>>> pkg.safe_count("absent_table") is None
True

Return a new :class:Table whose expression is predicate(expr).

The predicate receives the engine-native expression and returns a new engine-native expression (typically a filtered one). The caller writes the predicate in the engine’s own dialect — this method doesn’t translate between engines.

Parameters:

Returns:

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> expr = e.from_records([{"a": 1}, {"a": 2}, {"a": 3}])
>>> t = Table(name="t", expr=expr, engine=e)
>>> t2 = t.filter(lambda df: df.filter(df["a"] > 1))
>>> t2 is t
False
>>> t.count(), t2.count()
(3, 2)

Return a new :class:Table projected to the named columns.

Parameters:

Returns:

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> expr = e.from_records([{"a": 1, "b": 2}])
>>> Table(name="t", expr=expr, engine=e).select("a").columns()
['a']

Return a new :class:Table containing only the first n rows.

Lazy — the underlying engine builds a head expression, but no materialisation runs until the caller asks for it.

Parameters:

Returns:

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> expr = e.from_records([{"a": i} for i in range(10)])
>>> Table(name="t", expr=expr, engine=e).head(3).count()
3

Return the row count, pushing down to the engine where possible.

Delegates to :meth:Engine.count, which routes through each backend’s native cheap-count path (ibis pushes a SELECT COUNT(*) to duckdb; polars selects pl.len() inside the lazy plan; pandas calls len on the already- materialised frame). No table is materialised just to count it.

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> Table(name="t", expr=e.from_records([{"a": 1}]), engine=e).count()
1

Materialise as a pandas.DataFrame using the engine’s converter.

The returned frame uses the cross-engine nullable numpy-backed dtype family (Int64 / Float64 / string / boolean) regardless of which engine produced the expression — see :meth:corral.engines.base.Engine.to_pandas for the full contract.

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> t = Table(name="t", expr=e.from_records([{"a": 1}]), engine=e)
>>> t.to_pandas()["a"].tolist()
[1]

Materialise as a polars.DataFrame using the engine’s converter.

Raises :class:~corral.engines.errors.EngineNotAvailableError if polars is not installed and the engine cannot satisfy the conversion.

Examples:

>>> import pytest
>>> _ = pytest.importorskip("polars")
>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> t = Table(name="t", expr=e.from_records([{"a": 1}]), engine=e)
>>> t.to_polars().shape[0]
1

Force eager materialisation; return the engine-native frame.

For ibis this triggers backend execution and returns a pyarrow Table; for polars this collects the LazyFrame to a DataFrame; for pandas this is essentially an identity (pandas is already eager).

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> t = Table(name="t", expr=e.from_records([{"a": 1}]), engine=e)
>>> out = t.collect()
>>> out is not t
True

Return the column names of the underlying expression.

Delegates to :meth:Engine.columns, which uses the engine’s lazy schema introspection (ibis expr.schema().names, polars expr.collect_schema().names(), pandas list(expr.columns)). No row materialisation runs.

Examples:

>>> from corral.engines.ibis_engine import IbisEngine
>>> from corral.dataset import Table
>>> e = IbisEngine()
>>> Table(name="t", expr=e.from_records([{"a": 1, "b": 2}]), engine=e).columns()
['a', 'b']

Read a single CSV file into an engine-native lazy expression.

Adapters (CsvAdapter, ZipCsvAdapter after member extraction, RemoteAdapter after URL resolution) call this primitive directly. Engines should apply cast_schema at the tail when schema is non-None so behaviour matches the convenience scan(..., schema=...) path.

Parameters:

Read parquet (single file or Hive-partitioned directory).

When hive_partitioning=True the engine must enable partition discovery (so partition columns survive into the result and downstream filters become true partition prunes). For single-file parquet hive_partitioning is ignored.

Parameters:

Read one named table out of a .duckdb file.

Parameters:

Raises:

Build a table expression from in-memory records.

Accepts the two shapes a caller might naturally write:

• List of row dicts: [{"a": 1, "b": 2}, {"a": 3, "b": 4}]
• Columnar dict: {"a": [1, 3], "b": [2, 4]}

Adapters do not call this primitive (no on-disk format produces in-memory records); Engine.scan calls it directly for the {"data": ...} dict-source contract documented on :meth:scan.

Parameters:

Wrap a :class:pyarrow.Table as an engine-native lazy expression.

Type-preserving counterpart to :meth:from_records. Use this whenever the source is already an Arrow table — going through records = arrow.to_pylist() then from_records(records) is lossy (binary becomes bytes, decimals become Decimal, timestamps lose tz, all numerics widen via JSON-ish coercion) and is the round-trip the editing layer formerly used. Engines materialise the Arrow buffer directly:

• Ibis registers it as a duckdb temp table via con.create_table(name, obj=arrow_table, temp=True).
• Polars wraps it as a :class:polars.LazyFrame via pl.from_arrow(arrow_table).lazy().
• Pandas materialises with arrow_table.to_pandas().convert_dtypes() so the numpy-backed nullable dtype contract documented on :meth:to_pandas is preserved end-to-end.

Parameters:

Returns:

Examples:

>>> import pyarrow as pa
>>> from corral.engines.ibis_engine import IbisEngine
>>> tbl = pa.table({"a": [1, 2], "b": ["x", "y"]})
>>> e = IbisEngine()
>>> expr = e.from_arrow(tbl)
>>> list(expr.columns), expr.count().to_pyarrow().as_py()
(['a', 'b'], 2)

Write expr to dest as a single CSV file.

Parameters:

Write expr to dest as parquet (single file or partitioned).

When partition_by is given the engine writes a Hive-style partitioned dataset under dest. The parquet adapter currently handles partitioned writes through pyarrow regardless of engine; engines may either implement the partitioned-write path natively or document that they only support the single-file case.

Parameters:

Write expr into a .duckdb file as a named table.

Engines that cannot natively write duckdb without raw SQL (notably :class:~corral.engines.ibis_engine.IbisEngine) raise :class:~corral.engines.errors.EngineNotAvailableError and point callers at :class:~corral.engines.ibis_engine.IbisEngine for this primitive. The architecture’s “no raw SQL outside ibis_engine” rule is the constraint that forces this split.

Parameters:

Cast columns of expr per a Frictionless schema.

Fields whose Frictionless type does not map cleanly to an engine-native type are left untouched (the v0.3 apply_schema_to_df bug was a reminder that silent-but-noisy is the safer default for partial schemas). Fields named in the schema but absent from expr are skipped — that’s a validation concern, not a scan-time concern.

Parameters:

Returns:

Open source as a lazy table expression.

Convenience method — the body is a ~3-line delegation to corral.io.dispatch(source, format=format).read(source, engine=self, schema=schema, **kwargs). Power users and adapters skip this method entirely and call the primitives (:meth:read_csv, etc.) directly.

Parameters:

Returns:

Execute expr and return the engine’s native materialized frame.

For ibis this triggers backend execution; for polars this .collect()-s a LazyFrame; for pandas this is an identity (pandas is already eager).

Return the column names of expr without materialising rows.

Each engine has a cheap lazy schema path (ibis expr.schema().names, polars expr.collect_schema().names(), pandas list(expr.columns)). Implementations route to that path so the wrapper in :class:corral.dataset.Table is one line.

Parameters:

Returns:

Return the row count of expr, pushing the aggregate to the engine.

ibis pushes a SELECT COUNT(*) to duckdb; polars selects pl.len() inside the lazy plan; pandas calls len(expr). Implementations must not full-materialise the table to count it.

Parameters:

Returns:

Return a new lazy expression with only the first n rows.

Lazy — the engine builds a head expression but does not execute it. The pandas engine is the only one where this materialises (pandas has no lazy mode).

Parameters:

Returns:

Return a new lazy expression projected to columns.

Parameters:

Returns:

Return a new lazy expression sorted by columns.

ibis emits ORDER BY (expr.order_by); polars expr.sort; pandas sort_values (stable). descending applies to all columns. With :meth:limit this is the server-side paging path.

Parameters:

Returns:

Return a new lazy expression of n rows starting at offset.

ibis emits LIMIT n OFFSET offset; polars expr.slice; pandas positional iloc. Pairs with :meth:order_by for stable paging.

Parameters:

Returns:

Materialize expr and return it as a pandas.DataFrame.

Cross-engine convergence point. Implementations may raise :class:~corral.engines.errors.EngineNotAvailableError if pandas is not installed.

Dtype convention (cross-engine contract). The returned DataFrame uses pandas numpy-backed nullable dtypes: Int64 (capital I), Float64, string, boolean. We standardize on this family because:

• Null semantics are preserved without silently upcasting integer columns with missing values to float64 (the numpy default’s footgun).
• Numpy-backed (not pyarrow-backed) dtypes keep compatibility with downstream libraries (sklearn, older matplotlib, geopandas pre-1.0) that don’t understand pd.ArrowDtype columns.

Implementations achieve this by post-processing with :meth:pandas.DataFrame.convert_dtypes (the universal path), regardless of which engine produced expr. The Leavenworth fixture’s link.from_node_id column round-trips as Int64 from all three stock engines under this convention — that is the regression locked in by the cross-engine parity test.

Materialize expr and return it as a polars.DataFrame.

Cross-engine convergence point. Implementations may raise EngineNotAvailableError if polars is not installed.

Write expr to dest in format fmt.

Convenience method — the body is a ~3-line delegation to corral.io.get_adapter(fmt).write(expr, dest, engine=self, **kwargs). Adapters skip this and call the write primitives (:meth:write_csv, :meth:write_parquet, :meth:write_duckdb_table) directly.

Parameters:

Append an already-constructed :class:Issue to the report.

Parameters:

Examples:

>>> report = ValidationReport()
>>> report.add_issue(Issue(
...     severity=Severity.WARNING,
...     category=Category.SYNC_STATE,
...     code="sync.fk_stale",
...     message="link FK to node is stale",
... ))
>>> len(report.issues)
1

Construct an :class:Issue, append it, and return it.

This is the convenience builder validators usually call directly — the returned issue is handy when the validator wants to log or annotate the same finding elsewhere.

Parameters:

Returns:

Examples:

>>> report = ValidationReport()
>>> issue = report.add(
...     severity=Severity.ERROR,
...     category=Category.SCHEMA,
...     code="schema.required",
...     message="link.from_node_id row 0: value is null",
...     table="link",
...     column="from_node_id",
...     row=0,
... )
>>> issue.code
'schema.required'

Return all issues at the given severity, preserving insertion order.

Examples:

>>> r = ValidationReport()
>>> r.add(severity=Severity.ERROR, category=Category.SCHEMA,
...       code="schema.required", message="x")
Issue(severity=<Severity.ERROR: 'error'>, ...)
>>> len(r.by_severity(Severity.ERROR))
1
>>> r.by_severity(Severity.WARNING)
[]

Return all issues in the given category, preserving insertion order.

Examples:

>>> r = ValidationReport()
>>> r.add(severity=Severity.ERROR, category=Category.FOREIGN_KEY,
...       code="fk.missing_target", message="x")
Issue(...)
>>> [i.code for i in r.by_category(Category.FOREIGN_KEY)]
['fk.missing_target']

Return all issues attached to table, preserving insertion order.

Cross-cutting issues (Issue.table is None) are excluded — they would match every table query otherwise.

Examples:

>>> r = ValidationReport()
>>> r.add(severity=Severity.ERROR, category=Category.SCHEMA,
...       code="schema.required", message="x", table="link")
Issue(...)
>>> r.add(severity=Severity.ERROR, category=Category.STRUCTURAL,
...       code="structural.missing_table", message="y")  # no table
Issue(...)
>>> [i.table for i in r.by_table("link")]
['link']

Count issues, optionally filtered to a single severity.

Parameters:

Examples:

>>> r = ValidationReport()
>>> r.add(severity=Severity.ERROR, category=Category.SCHEMA,
...       code="schema.required", message="x")
Issue(...)
>>> r.add(severity=Severity.WARNING, category=Category.SYNC_STATE,
...       code="sync.fk_stale", message="y")
Issue(...)
>>> r.count()
2
>>> r.count(Severity.ERROR)
1

Return a JSON-serialisable dict snapshot of the report.

Schema:

.. code-block:: python

{
    "report_version": "1",
    "spec_version": "0.97" | None,
    "source": str | None,
    "created_at": "ISO-8601",
    "metadata": {...},
    "summary": {"error": N, "warning": N, "info": N,
                "data_quality": N, "is_clean": bool},
    "issues": [
        {"severity": "error", "category": "schema",
         "code": "...", "message": "...", "table": ..., ...},
        ...
    ],
}

Enum values flatten to their .value strings; datetime flattens via :meth:datetime.isoformat. The key set is stable; downstream consumers (the HTML renderer, the MCP server, the FastAPI server) rely on it.

Examples:

>>> r = ValidationReport(spec_version="0.97", source="x.gmns")
>>> d = r.to_dict()
>>> d["report_version"]
'1'
>>> d["summary"]["is_clean"]
True

Return the report as a JSON string. Optionally write to path.

Thin wrapper around :func:json.dumps on :meth:to_dict.

Parameters:

Examples:

>>> import json
>>> r = ValidationReport()
>>> data = json.loads(r.to_json())              # in-memory
>>> data["summary"]["is_clean"]
True
>>> import tempfile, pathlib
>>> with tempfile.TemporaryDirectory() as d:    # write-to-file
...     out = pathlib.Path(d) / "r.json"
...     _ = r.to_json(out)
...     out.is_file()
True

Return the rich-console rendering of this report.

Convenience wrapper around :func:corral.reports.render_rich. Kept here so that str(report) round-trips through the rich renderer without the caller importing :mod:corral.reports.render.

Examples:

>>> r = ValidationReport(source="empty.gmns")
>>> "empty.gmns" in r.to_rich()
True

Return the interactive single-file HTML rendering of this report.

Shortcut for :func:corral.reports.render_html. See its docstring for the offline-mode trade-off around the optional Vega-Lite map section.

Parameters:

Returns:

Examples:

>>> r = ValidationReport(source="empty.gmns")
>>> html = r.to_html()                              # in-memory
>>> html.lstrip().startswith("<!DOCTYPE html>")
True
>>> import tempfile, pathlib
>>> with tempfile.TemporaryDirectory() as d:        # write-to-file
...     out = pathlib.Path(d) / "r.html"
...     _ = r.to_html(out)
...     out.is_file()
True

Alias for :meth:to_rich — usable from print(report).

Hash on the stable identity fields, ignoring the mutable extra dict.

frozen=True normally generates __hash__, but the extra dict field is mutable and unhashable. Hashing on the identity fields (everything except extra) means equal issues — same severity, location, code, message, fix hint — collapse in a set(). This matches how a human would dedupe a report: two findings with the same code+message at the same row are the same finding, regardless of which validator’s debug payload they happen to carry.

Apply edit immediately, stamp with session_id, append to the log.

Raises :class:EditingError if called outside the with block.

Examples:

>>> from corral.dataset import Package, Table
>>> from corral.editing import Edit, Session
>>> from corral.engines.ibis_engine import IbisEngine
>>> e = IbisEngine()
>>> pkg = Package.from_tables({"x": Table(name="x", expr=e.from_records([{"a": 1}]), engine=e)})
>>> with Session(pkg) as s:
...     r = s.add_edit(Edit(op="add_rows", table="x", payload={"rows": [{"a": 9}]}))
>>> r.diff.rows_added
1

Reverse every edit applied in this session in LIFO order (without raising).

Cheap pre-check: should this rule run on this package at all?

Execute the rule; populate report with issues.

Return the estimated wall time in seconds.

The estimate is coefficient * (n_rows / 1_000_000) * n_tables where coefficient comes from :data:COEFFICIENTS (see :func:_resolve_coefficient for resolution rules). This is a documented heuristic, not a guarantee — see the module docstring.