Write Validation

Write Validation

Skill for composing effective data-validation plans with Pointblank. A validation plan is a sequence of steps that check different aspects of your data, organized to catch issues early and report clearly.

Quick start

import pointblank as pb

validation = (
    pb.Validate(
        data=df,
        tbl_name="orders",
        label="Order validation",
        thresholds=pb.Thresholds(warning=0.01, error=0.05),
    )
    # Structural checks first
    .col_exists(columns=["order_id", "amount", "status"])
    .row_count_match(count=1000, tol=50)

    # Value checks
    .col_vals_gt(columns="amount", value=0)
    .col_vals_not_null(columns="order_id")
    .col_vals_in_set(columns="status", set=["pending", "shipped", "delivered"])

    # Execute
    .interrogate()
)

Skill directory structure

skills/write-validation/
+-- SKILL.md                       <- This file
+-- references/
    +-- step-selection-guide.md    <- Which method for which concern
    +-- thresholds-and-actions.md  <- Configuring failure responses

When to use what

Core concepts

Ordering your steps

Organize validation steps from structural to semantic:

1. Schema checks -- col_exists, col_schema_match, col_count_match, row_count_match
2. Completeness checks -- col_vals_not_null, rows_complete, col_pct_null
3. Uniqueness checks -- rows_distinct
4. Value-range checks -- col_vals_gt, col_vals_between, etc.
5. Format checks -- col_vals_regex, col_vals_within_spec
6. Set membership -- col_vals_in_set, col_vals_not_in_set
7. Cross-column checks -- conjointly, col_vals_expr
8. Aggregate checks -- col_sum_eq, col_avg_gt
9. Freshness checks -- data_freshness
10. Custom/semantic checks -- specially, prompt

This order means structural problems surface first before value-level checks run.

Handling nulls

By default, null values count as failures in value checks. Control this per step:

# Nulls count as failures (default)
.col_vals_gt(columns="amount", value=0)

# Nulls are treated as passing
.col_vals_gt(columns="amount", value=0, na_pass=True)

For structured missingness (sentinel values like -999, "N/A"):

missing_spec = pb.MissingSpec(
    reasons={-999: "not collected", -1: "redacted"},
    null_is_missing=True,
    null_reason="unknown",
)

.col_vals_gt(columns="measurement", value=0, missing=missing_spec)
.col_pct_missing(columns="measurement", missing=missing_spec, max_pct=0.10)

Segmented validation

Break validation into groups to see which segments fail:

.col_vals_gt(
    columns="revenue",
    value=0,
    segments=pb.seg_group(["region_a", "region_b", "region_c"]),
)

Conditional steps

Skip steps dynamically based on the data:

# Only run if the column exists
.col_vals_gt(
    columns="new_feature",
    value=0,
    active=pb.has_columns("new_feature"),
)

# Only run if the table has enough rows
.rows_distinct(active=pb.has_rows(min=100))

Pre-processing data

Transform data before a check with pre:

.col_vals_gt(
    columns="price",
    value=0,
    pre=lambda df: df.filter(pl.col("status") == "active"),
)

Thresholds

Set thresholds at three severity levels:

# Global thresholds (apply to all steps)
pb.Validate(
    data=df,
    thresholds=pb.Thresholds(warning=0.01, error=0.05, critical=0.25),
)

# Per-step override
.col_vals_gt(
    columns="amount",
    value=0,
    thresholds=pb.Thresholds(warning=5, error=20),
)

• Values < 1: fraction of failing test units (e.g., 0.05 = 5%)
• Values >= 1: absolute count of failures (e.g., 5 = 5 rows)
• True: any failure triggers (equivalent to 1)

Actions

Trigger responses when thresholds are exceeded:

pb.Actions(
    warning="Warning: {col} step {step} at {time}",
    error=lambda: send_alert("Data quality error"),
    critical=[
        pb.send_slack_notification(webhook_url="https://..."),
        "Critical failure in {col}",
    ],
    highest_only=True,  # only fire the highest triggered level
)

Template variables: {type}, {level}, {step}, {col}, {val}, {time}.

Final actions

Run after all steps complete, with access to the full summary:

def check_overall(summary=None):
    summary = pb.get_validation_summary()
    if summary and summary["n_failed_steps"] > 0:
        send_report(summary)

pb.Validate(
    data=df,
    final_actions=pb.FinalActions(check_overall),
)

Comparing against reference data

Track drift by comparing current data to a reference table:

validation = (
    pb.Validate(data=current_df, reference=previous_df)
    .col_sum_eq(columns="revenue")    # value=None -> uses ref()
    .col_avg_eq(columns="quantity")
    .interrogate()
)

Extracting results

# Did everything pass?
validation.all_passed()

# Counts and fractions by step
validation.n_passed(i=1, scalar=True)
validation.f_failed(i=[1, 2, 3])

# Get failing rows for a step
extracts = validation.get_data_extracts(i=1, frame=True)

# Split into pass/fail subsets
pass_df = validation.get_sundered_data(type="pass")
fail_df = validation.get_sundered_data(type="fail")

# Machine-readable report
json_report = validation.get_json_report()

Saving and reloading

pb.write_file(validation, filename="daily_check.pb")
restored = pb.read_file("daily_check.pb")

Workflows

Building a validation plan from scratch

1. Profile the data with pb.DataScan(data=df) to understand distributions, nulls, and types.
2. Start with structural checks (col_exists, col_schema_match).
3. Add completeness checks (col_vals_not_null, rows_complete).
4. Add value constraints based on domain knowledge.
5. Set thresholds appropriate to the use case.
6. Run interrogate() and review the report.
7. Iterate: adjust thresholds, add missing checks, remove noisy ones.

Adding checks to an existing plan

Read the current validation code, identify uncovered columns or concerns, and add steps in the appropriate position (structural before semantic). Preserve the existing threshold/action configuration unless changing it is part of the task.

Diagnosing validation failures

1. Run interrogate() and check the report.
2. Use get_data_extracts(i=N, frame=True) to see failing rows.
3. Use get_step_report(i=N) for a detailed per-step view.
4. Determine whether the check or the data is wrong.
5. Fix the check (adjust value/threshold) or flag the data issue.

Gotchas

1. .interrogate() must be called. Steps are declarative until executed.
2. na_pass defaults to False. Nulls fail value checks unless you opt in.
3. Threshold 0.05 vs 5. Fractional = percentage, integer = count.
4. col_vals_between is closed by default. Pass inclusive=(False, False) for an open interval.
5. conjointly takes lambdas, not method calls. Each argument is lambda v: v.col_vals_gt(...).
6. Aggregate methods compare one value per column, not per row. They produce a single pass/fail per column.
7. segments splits the check into sub-groups. Each segment is reported separately in the validation report.