Skip to content

Validation Engine

Validation is performed through SchemaModel.validate(). The engine package exposes the result models the pipeline produces.

ValidationResult

nyctea.engine.results.ValidationResult dataclass

Result of a validation run.

Source code in src/nyctea/engine/results.py 
100
101
102
103
104
105
106
@dataclass(frozen=True)
class ValidationResult:
    """Result of a validation run."""

    data: pl.DataFrame | pl.LazyFrame
    errors: pl.DataFrame
    report: ValidationReport

ValidationReport

nyctea.engine.results.ValidationReport

Bases: BaseModel

Comprehensive validation outcome report.

Source code in src/nyctea/engine/results.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class ValidationReport(BaseModel):
    """Comprehensive validation outcome report."""

    model_config = ConfigDict(extra="forbid")

    rows_processed: int
    rows_valid: int
    on_failure: OnFailureBehavior
    columns: dict[str, ColumnValidationStats] = Field(default_factory=dict)

    def summary(self) -> str:
        """Return a human-readable summary."""
        pct = (self.rows_valid / self.rows_processed * 100) if self.rows_processed else 0.0
        lines = [
            f"Validation Report (on_failure: {self.on_failure})",
            f"Rows: {self.rows_valid}/{self.rows_processed} valid ({pct:.1f}%)",
            "",
            "Column Issues:",
        ]
        for col_name, stats in self.columns.items():
            if (
                stats.nullified > 0
                or stats.parse_failures > 0
                or stats.check_failures > 0
                or stats.coercion_failures > 0
            ):
                lines.append(f"  {col_name}:")
                if stats.parse_failures:
                    lines.append(f"    Parse failures: {stats.parse_failures}")
                if stats.coercion_failures:
                    lines.append(f"    Coercion failures: {stats.coercion_failures}")
                if stats.check_failures:
                    lines.append(f"    Check failures: {stats.check_failures}")
                if stats.nullified:
                    lines.append(f"    Nullified: {stats.nullified}")
                lines.append(f"    Final nulls: {stats.final_null_count}")
        return "\n".join(lines)

summary()

Return a human-readable summary.

Source code in src/nyctea/engine/results.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def summary(self) -> str:
    """Return a human-readable summary."""
    pct = (self.rows_valid / self.rows_processed * 100) if self.rows_processed else 0.0
    lines = [
        f"Validation Report (on_failure: {self.on_failure})",
        f"Rows: {self.rows_valid}/{self.rows_processed} valid ({pct:.1f}%)",
        "",
        "Column Issues:",
    ]
    for col_name, stats in self.columns.items():
        if (
            stats.nullified > 0
            or stats.parse_failures > 0
            or stats.check_failures > 0
            or stats.coercion_failures > 0
        ):
            lines.append(f"  {col_name}:")
            if stats.parse_failures:
                lines.append(f"    Parse failures: {stats.parse_failures}")
            if stats.coercion_failures:
                lines.append(f"    Coercion failures: {stats.coercion_failures}")
            if stats.check_failures:
                lines.append(f"    Check failures: {stats.check_failures}")
            if stats.nullified:
                lines.append(f"    Nullified: {stats.nullified}")
            lines.append(f"    Final nulls: {stats.final_null_count}")
    return "\n".join(lines)

ColumnValidationStats

nyctea.engine.results.ColumnValidationStats

Bases: BaseModel

Per-column validation statistics.

Source code in src/nyctea/engine/results.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class ColumnValidationStats(BaseModel):
    """Per-column validation statistics."""

    model_config = ConfigDict(extra="forbid")

    column_name: str
    parse_failures: int = 0
    coercion_failures: int = 0
    check_failures: int = 0
    nullified: int = Field(0, description="Values set to null due to failures")
    final_null_count: int = Field(0, description="Total nulls in output")
    original_null_count: int = Field(
        0,
        description="Nulls after column resolution and before frame and column transformations",
    )

ErrorReportConfig

nyctea.engine.results.ErrorReportConfig

Bases: BaseModel

Configuration for error reporting detail level.

Source code in src/nyctea/engine/results.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class ErrorReportConfig(BaseModel):
    """Configuration for error reporting detail level."""

    model_config = ConfigDict(extra="forbid")

    mode: Literal["summary", "rows", "cells"] = Field(
        "summary",
        description=(
            "Error reporting mode:\n"
            "- 'summary': Column + check + count only (minimal)\n"
            "- 'rows': Add row indices where failures occurred\n"
            "- 'cells': Add row indices + actual values (maximum detail)"
        ),
    )
    limit: int | None = Field(
        None,
        ge=0,
        description="Maximum number of error rows to return per column+check. None = unlimited.",
    )
    include_values: bool = Field(
        True,
        description="Include actual failing values in output (only applies to 'cells' mode)",
    )