Validation Definitions and Checkpoints

46.1 Validation Definitions (23 min)

So far you've validated by fetching a batch and calling batch.validate(suite) yourself. A validation definition saves that pairing (this data + this suite) as a named, reusable object in the project.

import great_expectations as gx
from gx_helpers import get_context

context = get_context()
ds = context.data_sources.get("macro_duckdb")

batch_def = ds.get_asset("macro_monthly").get_batch_definition("full")
suite = context.suites.get("macro_monthly__critical")

vd = context.validation_definitions.add(
    gx.ValidationDefinition(name="macro_monthly__critical__full", data=batch_def, suite=suite)
)
result = vd.run()
print(vd.name, "→", result.success)

It's stored in quality/gx/validation_definitions/, so any script or scheduler can run it by name:

vd = context.validation_definitions.get("macro_monthly__critical__full")
print(vd.run().success)

For a monthly batch definition, pass batch_parameters to choose a month; omit them for the latest:

context.validation_definitions.get("stg_statcan__cpi__critical__by_month").run(
    batch_parameters={"year": 2026, "month": 7})

46.2 Checkpoints and Actions (28 min)

A checkpoint runs one or more validation definitions and then does something with the results, through actions.

Checkpoint "critical"
 ├── validation definitions:  every <asset>__critical__<batch>
 ├── result_format:           SUMMARY
 └── actions:
      ├── UpdateDataDocsAction      → rebuild the Data Docs website (Day 47)
      └── (optional) notification   → Slack / Teams / e-mail on failure

Creating checkpoints

import great_expectations as gx

critical_vds = [vd for vd in context.validation_definitions.all()
                if "__critical__" in vd.name]

checkpoint = context.checkpoints.add(gx.Checkpoint(
    name="critical",
    validation_definitions=critical_vds,
    actions=[gx.checkpoints.UpdateDataDocsAction(name="update_data_docs")],
    result_format={"result_format": "SUMMARY"},
))

result = checkpoint.run()
print("Critical checkpoint success:", result.success)

Depending on your GX 1.x release, the actions module may be gx.checkpoint or gx.checkpoints. If one import fails, try the other: from great_expectations.checkpoint import UpdateDataDocsAction.

Why two checkpoints?

This turns the severity column of your Day 41 plan into pipeline behaviour.

Result formats at checkpoint level

result_format={"result_format": "SUMMARY"} keeps results informative but compact. Use COMPLETE only when debugging a specific failure, because on large tables it can store many unexpected values.

46.3 Extending the Builder (28 min)

Upgrade quality/build_suites.py so that one command regenerates everything from the specs, in a safe order:

delete:  checkpoints → validation definitions → suites     (things that reference others go first)
create:  suites → validation definitions → checkpoints

Replace main() and add helpers:

import great_expectations as gx


def suite_name_for(spec, severity):
    """<asset>__<severity>, plus __<batch> when the spec targets a non-default batch."""
    batch = spec.get("batch_definition", "full")
    suffix = "" if batch == "full" else f"__{batch}"
    return f"{spec['asset']}__{severity}{suffix}"


def clear_all(context):
    for cp in list(context.checkpoints.all()):
        context.checkpoints.delete(name=cp.name)
    for vd in list(context.validation_definitions.all()):
        context.validation_definitions.delete(name=vd.name)
    for s in list(context.suites.all()):
        context.suites.delete(name=s.name)


def main():
    context = get_context()
    ds = context.data_sources.get("macro_duckdb")
    clear_all(context)

    vds = {"critical": [], "warning": []}
    for spec_name, spec in load_specs():
        batch_name = spec.get("batch_definition", "full")
        batch_def = ds.get_asset(spec["asset"]).get_batch_definition(batch_name)
        for severity in SEVERITIES:
            items = spec.get(severity) or []
            if not items:
                continue
            s_name = suite_name_for(spec, severity)
            suite = rebuild_suite(context, s_name, items, spec.get("owner"))
            vd = context.validation_definitions.add(gx.ValidationDefinition(
                name=f"{s_name}__{batch_name}" if batch_name == "full" else s_name,
                data=batch_def, suite=suite))
            vds[severity].append(vd)
            print(f"✓ {vd.name:<60} {len(items)} expectation(s)")

    for severity, defs in vds.items():
        if defs:
            context.checkpoints.add(gx.Checkpoint(
                name=severity,
                validation_definitions=defs,
                actions=[gx.checkpoints.UpdateDataDocsAction(name="update_data_docs")],
                result_format={"result_format": "SUMMARY"},
            ))
            print(f"✓ checkpoint '{severity}' with {len(defs)} validation definition(s)")

python quality/build_suites.py
ls quality/gx/checkpoints quality/gx/validation_definitions

Everything GX runs is now derived from the specs. Changing a threshold means editing YAML, reviewing the diff, and re-running the builder.

46.4 Running Checkpoints and Interpreting Results (20 min)

from gx_helpers import get_context

context = get_context()
result = context.checkpoints.get("critical").run()

print("Checkpoint success:", result.success)
for identifier, vr in result.run_results.items():
    stats = vr.statistics
    print(f"{'✓' if vr.success else '✗'} {vr.suite_name:<55} "
          f"{stats['successful_expectations']}/{stats['evaluated_expectations']}")
    for r in vr.results:
        if not r.success:
            cfg = r.expectation_config
            meta = cfg.meta or {}
            print(f"     ✗ [{meta.get('plan_id')}] {cfg.type} {cfg.kwargs.get('column', '')}"
                  f" → {r.result.get('unexpected_count', r.result.get('observed_value'))}"
                  f" | {meta.get('rationale', '')[:80]}")

result.run_results is a dictionary: one entry per validation definition, each holding the suite-level result you met on Day 44. Printing the rationale next to each failure turns a cryptic red cross into an actionable message.

Try it 46.4

Temporarily edit specs/fct_open_economy_daily.yml so that one critical expectation must fail (e.g. usd_cad max 1.2). Rebuild, run the critical checkpoint, and check that your report names the plan ID and rationale. Then restore the spec and rebuild. This is the "break it on purpose" habit from Day 29.

46.5 Notifications (15 min)

GX includes notification actions (for example SlackNotificationAction and MicrosoftTeamsNotificationAction) that post a summary when a checkpoint fails:

from great_expectations.checkpoint import SlackNotificationAction

notify = SlackNotificationAction(
    name="notify_on_failure",
    slack_webhook="${SLACK_WEBHOOK_URL}",       # substituted from env / config_variables.yml
    notify_on="failure",
    show_failed_expectations=True,
)

Add it to a checkpoint's actions list. Never write the webhook URL itself into code or YAML: it's a secret (Day 23).

Our choice: we already have a pipeline-wide notify() helper and alert policy from Day 29. To keep one alerting channel with one set of rules (and avoid alert fatigue), run_checkpoints.py builds its own message from the results and calls a small notifier using the same NOTIFICATIONS__WEBHOOK_URL. GX's built-in actions are a good alternative when GX runs on its own.

46.6 Mini-Project: run_checkpoints.py with Exit Codes (23 min)

Create quality/run_checkpoints.py:

1. Runs the critical checkpoint, then the warning checkpoint.
2. Prints a compact report: per suite, passed/evaluated; per failure, plan ID, expectation, column, detail, rationale.
3. Sends one alert message if anything failed (critical first).
4. Exits with 0 = all passed, 1 = a critical failure (do not publish), 2 = warnings only (publish, review).

These exit codes follow the same convention as ingest.py (Day 30).

"""Run critical and warning checkpoints, report, alert, and exit with a meaningful code.

Exit codes: 0 all passed · 1 critical failure · 2 warnings only
"""
import os
import sys

import requests

from gx_helpers import get_context


def summarise(checkpoint_result):
    lines, failed = [], 0
    for vr in checkpoint_result.run_results.values():
        s = vr.statistics
        lines.append(f"  {'✓' if vr.success else '✗'} {vr.suite_name}: "
                     f"{s['successful_expectations']}/{s['evaluated_expectations']}")
        for r in vr.results:
            if r.success:
                continue
            failed += 1
            cfg, meta = r.expectation_config, (r.expectation_config.meta or {})
            detail = r.result.get("unexpected_count", r.result.get("observed_value"))
            lines.append(f"      ✗ [{meta.get('plan_id', '-')}] {cfg.type} "
                         f"{cfg.kwargs.get('column', '')} → {detail}")
            if meta.get("rationale"):
                lines.append(f"        why it matters: {meta['rationale'].strip()[:120]}")
    return failed, lines


def notify(message):
    url = os.environ.get("NOTIFICATIONS__WEBHOOK_URL")
    if not url:
        print(f"[ALERT]\n{message}")
        return
    try:
        requests.post(url, json={"text": message}, timeout=10)
    except Exception as e:
        print(f"[ALERT not delivered: {e}]\n{message}")


def main():
    context = get_context()
    report, outcome = [], {}

    for name in ("critical", "warning"):
        try:
            result = context.checkpoints.get(name).run()
        except Exception as e:
            report.append(f"✗ checkpoint '{name}' could not run: {e}")
            outcome[name] = 1
            continue
        failed, lines = summarise(result)
        outcome[name] = failed
        report.append(f"{'✓' if failed == 0 else '✗'} checkpoint '{name}': "
                      f"{failed} failing expectation(s)")
        report.extend(lines)

    text = "\n".join(report)
    print(text)

    if outcome.get("critical"):
        notify("⛔ Data quality: CRITICAL failures — marts should not be published.\n" + text)
        sys.exit(1)
    if outcome.get("warning"):
        notify("⚠ Data quality: warnings need review.\n" + text)
        sys.exit(2)
    print("\n✓ All data quality checks passed")


if __name__ == "__main__":
    main()

python quality/build_suites.py
python quality/run_checkpoints.py; echo "exit code: $?"

Day 46 checklist

• I can create, save and run validation definitions, for whole tables and specific months
• Separate critical and warning checkpoints turn plan severities into pipeline behaviour
• One builder regenerates suites, definitions and checkpoints from specs in a safe order
• I can interpret checkpoint results and report failures with plan IDs and rationales
• Alerts go through one channel; exit codes are 0 / 1 (critical) / 2 (warnings)

git add quality/ && git commit -m "Day 46: validation definitions, checkpoints, run_checkpoints" && git push