Validation

The validation module checks data quality against a set of rules and produces a structured ValidationResult. Rules observe data and report whether it satisfies conditions; they never modify the data.

validate()

The primary entry point:


import polars as pl
from hermes.validation import validate, NotNull, RangeCheck
 
data = pl.read_csv("data.csv")
 
report = validate(
    data=data,
    rules=[
        NotNull("country_code"),
        RangeCheck("value", min=0, max=100),
    ],
)
 
print(report.passed)     # True if every rule passed
print(report.summary())  # human-readable summary

Parameters

Parameter	Type	Description
`data`	`Any`	`pl.DataFrame`, `pl.LazyFrame`, `pyarrow.Table`, or dict/list of dicts
`rules`	`list[ValidationRule]	None`
`context`	`ValidationContext	None`

Rules run independently against the original input — they never feed results to each other.

ValidationResult

Field	Type	Description
`success`	`bool`	The validation process completed without engine errors
`passed`	`bool`	`success` and every rule passed
`results`	`list[RuleResult]`	Per-rule outcomes
`errors`	`list[str]`	Engine errors (e.g. a rule raised)
`warnings`	`list[str]`	Rule warnings

Methods: summary() (formatted text), to_dict() (JSON-serializable), bool(report) is passed.

RuleResult

Field	Type	Description
`rule`	`str`	Rule name
`passed`	`bool`	Whether the rule passed
`statistics`	`dict`	Check-specific stats (e.g. null counts, violation counts)
`violations`	`list[Violation]`	Offending values
`message`	`str`	Human-readable result
`warnings`	`list[str]`	Warnings

Violation

Violation(field, row=None, value=None, expected=None, message="") — a single offending value.

ValidationEngine

For running a configured rule pipeline:


from hermes.validation import ValidationEngine, NotNull
 
engine = ValidationEngine(rules=[NotNull("id")])
engine.add_rule(RangeCheck("value", min=0))
 
result = engine.validate(data)
engine.rules                       # list of configured rules
engine.describe()                  # rule metadata as dicts

Built-in Rules

All rules are exported from hermes.validation. Each exposes name, check(data, context) -> RuleResult, validate() (raises RuleConfigurationError if misconfigured), and describe(). Column-based rules take a field argument (the column name).

Rule	Constructor	Checks
`NotNull`	`NotNull("col")`	Column has no nulls
`Unique`	`Unique("col")`	Column values are unique
`UniqueCombination`	`UniqueCombination(["a", "b"])`	Row tuples are unique
`TypeCheck`	`TypeCheck("col", "float")` or `TypeCheck({"a": "float"})`	Column schema types
`RangeCheck`	`RangeCheck("col", min=0, max=100)`	Numeric values within range
`EnumCheck`	`EnumCheck("col", ["US", "GB"])`	Values are in an allowed set
`RegexCheck`	`RegexCheck("col", r"\d{4}")`	Values match a regex
`PatternCheck`	`PatternCheck("col", "pattern")`	Named pattern match
`LengthCheck`	`LengthCheck("col", min=..., max=...)`	String lengths
`DateRangeCheck`	`DateRangeCheck("date", min=..., max=...)`	Dates within range
`DateOrderCheck`	`DateOrderCheck("start", "end")`	`start` is not later than `end`
`SchemaCheck`	`SchemaCheck({"col": "type"})`	Column names/types match exactly
`ColumnCheck`	`ColumnCheck(["a", "b"], allow_extra=True)`	Required columns present
`RowCountCheck`	`RowCountCheck(min=None, max=None, exact=None)`	Row count bounds
`NullRateCheck`	`NullRateCheck("col", max_rate=0.1)`	Null ratio below threshold
`DuplicateCheck`	`DuplicateCheck(columns=None)`	Duplicate rows (whole row or subset)
`CompletenessCheck`	`CompletenessCheck(["a"], min_rate=1.0)`	Combined completeness of columns
`FreshnessCheck`	`FreshnessCheck("date", max_age="7d")`	Most recent value is not stale
`ReferentialCheck`	`ReferentialCheck("col", reference_data=...)`	Values exist in a reference dataset
`ForeignKeyCheck`	`ForeignKeyCheck("col", reference=...)`	Values match a reference column
`ConstantCheck`	`ConstantCheck("col", expected=None)`	Column is constant
`CardinalityCheck`	`CardinalityCheck("col", unique_min=..., unique_max=...)`	Distinct-count bounds

register_pattern(name, compiled_regex) registers named patterns used by PatternCheck.

Top-Level API


import hermes as hr
from hermes.validation import NotNull
 
report = hr.validate(df, rules=[NotNull("country_code")])

Errors


from hermes.validation import (
    ValidationError,          # base error
    RuleConfigurationError,   # a rule is misconfigured
    RuleExecutionError,       # a rule failed while running
)