Features¶

An overview of what Nyctea provides.

Lazy by default¶

All pipeline phases operate on Polars LazyFrame. The data is never collected inside the validator unless you set lazy=False. Error reporting uses targeted collects of only the columns it needs.

result = schema.validate(df, registry)        # result.data is a LazyFrame
result = schema.validate(df, registry, lazy=False)  # result.data is a DataFrame

Streaming engine for internal aggregates¶

Validation's internal aggregate collects (check and coercion enforcement, summary error counts, report building) are pure reductions: sums, lengths, boolean all(). Above nyctea.Config.streaming_row_threshold() rows, they use Polars' streaming engine, which cuts peak memory substantially on large data. Below it, they use the default engine, since streaming has a fixed per-query setup cost that outweighs the reduction itself on small data. The default threshold is 100,000 rows, based on a measured crossover. Override it if your workload sits far from that. It describes the machine and the run rather than what valid data looks like, so it lives on nyctea.Config, not in a schema file:

import nyctea

nyctea.Config.set_streaming_row_threshold(10_000)  # lower it if you mostly validate large files

with nyctea.Config(streaming_row_threshold=10_000):  # or scope it to one block
    result = schema.validate(df, registry)

A LazyFrame input always uses streaming, since its row count isn't known without collecting, and passing lazy input in the first place signals larger or out-of-core data.

The rows and cells error report modes are the exception. They materialise row indices and failing values rather than reducing them, so the threshold does not apply to them and they pass no engine= at all. That leaves them on Polars' own default selection, engine="auto", which means they do follow your global affinity setting where the aggregates below deliberately do not. Only the summary mode's error counts are an aggregate.

These aggregate collects always pass an explicit engine= value, so they don't respect pl.Config.set_engine_affinity() or the POLARS_ENGINE_AFFINITY environment variable, unlike engine="auto" calls elsewhere in your own code. That is deliberate. streaming_row_threshold is a per-query decision that knows how much data it is looking at, while Polars' engine affinity is a single global switch with no size awareness (see the engine parameter on LazyFrame.collect). If your global affinity is set to "streaming", nyctea's internal aggregates on small data still use the in-memory engine unless you lower streaming_row_threshold yourself. The rows and cells collects described above are the ones that will follow it, since they leave the choice to Polars.

Composable pipeline¶

Validation runs through ordered phases. Each phase receives a PipelineContext and returns it with updated state.

Phases can be added, removed, or reordered, including CoercionPhase relative to parsing and checks: a check written against raw strings can run before coercion, one written against typed values can run after. Two positions are fixed. ColumnResolutionPhase always runs first, since every other phase reads resolved names. NotNullPhase always runs last, since it answers "does the output contain nulls", which only means something once everything that could introduce one has already run. ValidationPipeline rejects any ordering that violates either.

Validator registry¶

Parsers and checks are registered by name in a Registry. One registry can be shared across many schemas.

from nyctea import Registry, register_builtins

registry = Registry()
register_builtins(registry)  # strip, lower, upper, to_int, to_float, min_value, between, in_set, unique

Custom validators can be added via OOP classes or the decorator API:

from nyctea import checker, frame_checker, frame_parser, parser


@checker(registry=registry, name="positive", description="Value must be > 0")
def positive(column: pl.Expr) -> pl.Expr:
    return column > 0

Schema definition¶

Schemas are defined as Python dicts or YAML. Each column specifies a dtype, nullability, parsers, and checks.

schema = SchemaModel.from_dict({
    "coerce": True,
    "on_failure": "null",
    "columns": {
        "age": {
            "dtype": "Int64",
            "nullable": True,
            "parsers": [{"name": "strip"}, {"name": "to_int"}],
            "checks": [{"name": "min_value", "args": {"min": 0}}],
        },
        "name": {
            "dtype": "Utf8",
            "nullable": False,
            "synonyms": ["Name", "NAME"],
        },
    }
})

Frame-level parsers and checks¶

Column parsers and checks operate on one column at a time. Frame parsers and checks operate on the whole DataFrame, for rules that need to see more than one column, such as cross-column comparisons or a minimum row count.

from nyctea import checker, frame_checker, frame_parser, parser


@frame_checker(registry=registry, name="min_rows")
def min_rows(frame: pl.LazyFrame, min_rows: int = 1) -> pl.LazyFrame:
    if frame.select(pl.len()).collect().item() < min_rows:
        raise ValueError(f"expected at least {min_rows} rows")
    return frame

schema = SchemaModel.from_dict({
    "frame_checks": [{"name": "min_rows", "args": {"min_rows": 2}}],
    "columns": {"age": {"dtype": "Int64"}},
})

Frame checks raise on failure. Whatever a check returns is discarded, so it cannot change the data. Frame parsers may add, drop, or reorder columns and rows, and run before column parsers so later steps see the transformed frame.

Failure handling¶

The on_failure field controls what happens when validation fails. Set it at schema level or per column.

Per-column settings override the schema default:

{
    "on_failure": "null",
    "columns": {
        "id": {"dtype": "Int64", "nullable": False, "on_failure": "raise"},
        "score": {"dtype": "Float64", "nullable": True},  # inherits "null"
    }
}

on_failure: "null" is legal on a non-nullable column. A resulting not-null violation is reported rather than raised, the same way on_failure: "ignore" behaves.

Which exception you get¶

schema.validate raises three kinds of error, and the distinction is about whose problem it is.

ConfigurationError means the schema itself does not hold up. A check or parser it names is not in the registry, an argument does not bind, or a column declares the same check twice. This is checked before any data is read, so it comes first and does not depend on the data at all.

ValidationError means your input does not have the structure the schema describes. A required column is missing from the input, or a name resolves ambiguously because both the canonical name and a synonym are present. Nyctea cannot start validating, because it cannot tell which column is which. This covers resolving the columns you passed in. A frame parser that removes a required column mid-run is reported as PipelineError, because by then validation had started.

PipelineError covers the rest of a validation run. That means a check, parser, coercion or nullability failure on a column set to on_failure: "raise", and a phase that failed while building its part of the query. When a phase fails unexpectedly, the original exception is kept as the __cause__.

ValidationError and PipelineError carry a phase attribute, and a column whenever one column owns the failure. Expect column to be None for a failure that belongs to the frame rather than to any single column, such as a frame parser removing required columns. ConfigurationError carries neither, because it is raised against the schema before any phase has run. A PipelineError raised before the first phase, such as a reserved internal column already present in your data, also has phase set to None.

from nyctea import ConfigurationError, PipelineError, ValidationError

try:
    result = schema.validate(df, registry)
except ConfigurationError as e:
    print(f"Schema does not verify against the registry: {e}")
except ValidationError as e:
    print(f"Schema does not fit the input: column {e.column!r} in phase {e.phase!r}")
except PipelineError as e:
    print(f"Validation stopped: column {e.column!r} in phase {e.phase!r}")

All three inherit from NycteaError, so catch that to handle any of them.

Nyctea builds a lazy query and evaluates it in one pass, so a check or parser whose expression builds correctly but fails on the data fails after every phase has run. That surfaces as PipelineError with the underlying Polars exception as its __cause__, and with no phase, because no phase was running. A check that casts with strict=True against a value that will not fit is one way to get there. Prefer checks that return a boolean expression over the column and leave failure handling to on_failure, rather than ones that fail on bad data.

A custom phase can raise ValidationError itself to report a structural problem of its own. The pipeline passes it through untouched rather than wrapping it, which is how a phase you write says "this data cannot be validated" instead of "this phase broke". Everything else a phase raises, from execute or from the can_skip and can_change_row_count hooks, is wrapped as PipelineError with the original kept as __cause__. That includes a PipelineError the phase raised itself. The wrapper sets phase to the phase that was running, so a phase cannot attribute its failure elsewhere, and carries across the column the phase set.

Parser failures and null counts¶

A parser failure occurs when a non-null value becomes null anywhere across a column's parser chain. Nyctea reports it under the reserved check name parse, counts it in report.columns[column].parse_failures, and marks the row invalid. Input values that were already null are not parser failures.

original_null_count counts nulls after synonyms are resolved but before frame parsers and column parsers run. final_null_count counts nulls in the validated output. A frame parser that filters rows can therefore make the original count larger than the final row count.

Error reporting¶

ErrorReportConfig controls the detail level of the errors DataFrame returned by validate().

Summary mode (default)¶

One row per failing check with a count.

column | check     | count
age    | min_value | 5

Rows mode¶

Adds a list of failing row indices.

column | check     | count | row_indices
age    | min_value | 5     | [0, 3, 7, 12, 15]

Cells mode¶

One row per failing cell with the actual value.

column | check     | row_index | value
age    | min_value | 0         | -5
age    | min_value | 3         | -1

A check answers one question per row, so its expression evaluates to one boolean per row. An expression that aggregates to a single answer for the whole frame is rejected, because it cannot be reported per row or nulled per value. That is detected by comparing lengths, so it needs at least two rows to tell apart: on a single row the two are the same length. Combining an aggregate with a per-row operand is fine, so column - column.min() < 2 and column.count().over(column) >= 3 both work.

row_index is a position in result.data, not in the frame you passed in. The two agree unless a frame parser reordered or removed rows, since validation assigns positions again after frame parsing.

Use limit to cap the number of error entries per column+check:

config = ErrorReportConfig(mode="cells", limit=100)
result = schema.validate(df, registry, error_report_config=config)

The value column renders each failing value as text. A Binary column is hex-encoded rather than decoded, because bytes that are not valid UTF-8 have no text form. That applies to every Binary column, so a value's representation does not depend on the bytes it happens to hold. Nested dtypes are the exception: a List, Array or Object column has no text rendering, so include_values is not supported for one and the report fails.

Coercion¶

When coerce=True, columns are cast to their target dtype before checks run. Failed casts are handled according to on_failure.

Per-column overrides are supported:

{
    "coerce": False,
    "columns": {
        "age": {"dtype": "Int64", "coerce": True},   # coerced
        "name": {"dtype": "Utf8"},                     # not coerced
    }
}

Column synonyms¶

Columns can have synonyms for automatic renaming. If a synonym matches a column in the input data, it is renamed to the canonical name.

"name": {"dtype": "Utf8", "synonyms": ["Name", "NAME", "full_name"]}

Ambiguous matches (both canonical and synonym present) raise ValidationError from the column resolution phase.

Cleaned column matching¶

Exact matching is the default and is unchanged. Set column_matching: cleaned on the schema to also accept names that match after trimming whitespace and Unicode case folding, so AGE resolves to age without enumerating every casing as a synonym.

column_matching: cleaned
columns:
  age:
    dtype: Int64

Exact matches always win over cleaned ones. If two physical columns match one schema column at the cleaned level, that is ambiguous and resolution fails rather than guessing. A schema whose accepted names collide once cleaned, such as a column age and a synonym AGE on a different column, is rejected when it is constructed.